From ac0566f779827bbd1d596ec77c6d3a2087bb1ddc Mon Sep 17 00:00:00 2001 From: iamxvbaba <28732408+iamxvbaba@users.noreply.github.com> Date: Sun, 2 Aug 2026 01:51:03 +0800 Subject: [PATCH] refactor(mtproto): sync redesign RPC replay ownership --- .env.example | 6 +- cmd/telesrv/main.go | 3 - docs/configuration.en.md | 11 +- docs/configuration.zh-CN.md | 13 +- internal/config/config.go | 67 +++-- internal/config/config_test.go | 54 ++-- internal/mtprotoedge/conn.go | 6 +- internal/mtprotoedge/encrypted.go | 41 ++- internal/mtprotoedge/inbound_layer_rpc.go | 17 +- .../mtprotoedge/inbound_layer_rpc_test.go | 133 ++++++++- internal/mtprotoedge/inbound_preflight.go | 23 +- .../layer_admission_budget_test.go | 67 +---- .../mtprotoedge/layer_rpc_execution_test.go | 10 +- internal/mtprotoedge/logical_session.go | 186 ++++++++++++ internal/mtprotoedge/logical_session_test.go | 281 ++++++++++++++++++ internal/mtprotoedge/outbound.go | 255 ++++++++++++---- internal/mtprotoedge/rpc_capacity_error.go | 20 ++ internal/mtprotoedge/rpc_flight.go | 50 +++- internal/mtprotoedge/rpc_result_cache.go | 153 ++++++++-- internal/mtprotoedge/rpc_result_cache_test.go | 36 +-- .../mtprotoedge/rpc_result_delivery_test.go | 14 +- .../rpc_result_retention_budget_test.go | 20 +- internal/mtprotoedge/rpc_rewrap.go | 66 ++-- internal/mtprotoedge/rpc_rewrap_test.go | 12 +- internal/mtprotoedge/rpc_test.go | 6 +- internal/mtprotoedge/server.go | 55 ++-- internal/mtprotoedge/session_boundary_test.go | 2 +- internal/mtprotoedge/session_manager.go | 40 +++ 28 files changed, 1304 insertions(+), 343 deletions(-) create mode 100644 internal/mtprotoedge/logical_session.go create mode 100644 internal/mtprotoedge/logical_session_test.go create mode 100644 internal/mtprotoedge/rpc_capacity_error.go diff --git a/.env.example b/.env.example index 48aaa223..baebc463 100644 --- a/.env.example +++ b/.env.example @@ -31,13 +31,11 @@ TELESRV_MTPROTO_RPC_TIMEOUT=30s TELESRV_MTPROTO_RPC_GLOBAL_WORKERS=256 TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS=8192 TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES=536870912 -# In-process 331s rpc_result ownership budgets: global >= auth >= session. +# Metadata-only rpc_result receipt budgets: global >= auth >= session. ACK deletes immediately; +# 331s is only the no-ACK horizon. Payloads live solely in the logical-session outbound budget. TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES=262144 -TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_BYTES=67108864 TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES=32768 -TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES=33554432 TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES=16384 -TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES=16777216 TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH=2048 # Process-wide in-flight transport wire + decrypted plaintext reservation. TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES=536870912 diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index b0d50523..ece41885 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -1417,11 +1417,8 @@ func run(logger *zap.Logger) error { RPCGlobalMaxTasks: cfg.MTProtoRPCGlobalMaxTasks, RPCGlobalMaxBytes: cfg.MTProtoRPCGlobalMaxBytes, RPCResultCacheMaxEntries: cfg.MTProtoRPCResultCacheMaxEntries, - RPCResultCacheMaxBytes: cfg.MTProtoRPCResultCacheMaxBytes, RPCResultCacheAuthMaxEntries: cfg.MTProtoRPCResultCacheAuthMaxEntries, - RPCResultCacheAuthMaxBytes: cfg.MTProtoRPCResultCacheAuthMaxBytes, RPCResultCacheSessionMaxEntries: cfg.MTProtoRPCResultCacheSessionMaxEntries, - RPCResultCacheSessionMaxBytes: cfg.MTProtoRPCResultCacheSessionMaxBytes, RPCResultPendingPerAuth: cfg.MTProtoRPCResultPendingPerAuth, InboundFrameGlobalMaxBytes: cfg.MTProtoInboundFrameGlobalMaxBytes, OutboundQueueSize: cfg.MTProtoOutboundQueueSize, diff --git a/docs/configuration.en.md b/docs/configuration.en.md index 8e95ce0c..8da196d0 100644 --- a/docs/configuration.en.md +++ b/docs/configuration.en.md @@ -35,17 +35,14 @@ This document describes every setting loaded by `internal/config`. Defaults and | `TELESRV_MTPROTO_RPC_GLOBAL_WORKERS` | int / `256` | Shared fair-scheduler worker count. | | `TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS` | int / `8192` | Process-wide scheduled/in-flight RPC task cap. | | `TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES` | int64 charge bytes / `536870912` | Process-wide reserved/queued/in-flight RPC memory charge. Exact admission reserves a conservative typed-materialization charge from wire size and grows it atomically before a nested-gzip-expanded graph is decoded; grow failure rejects the complete candidate batch. This is not an equal amount of concurrently receivable wire bytes. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES` | int / `262144` | Global ownership entries for pending owners, completed results, and tombstones during the in-process 331-second replay window. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_BYTES` | int64 bytes / `67108864` | Global retained-byte budget. Owner admission reserves one byte; Put transfers it to a body or tombstone. Must be at least `16775168`. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES` | int / `32768` | Per raw-auth-key ownership entries; charged together with global and session scopes. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES` | int64 bytes / `33554432` | Per raw-auth-key retained bytes. Limits must satisfy `global >= auth >= session`. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES` | int / `16384` | Per `raw auth key + session_id` ownership entries. | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES` | int64 bytes / `16777216` | Per `raw auth key + session_id` retained bytes; large enough for one legal outbound body. | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES` | int / `262144` | Global cap for pending owners and compact unacknowledged-result receipts. Receipts retain request identity, execution outcome, and Layer admission metadata only—never TL bodies. `msgs_ack` removes them immediately; 331 seconds is only the no-ACK safety horizon. | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES` | int / `32768` | Per-raw-auth owner/receipt cap; limits satisfy `global >= auth >= session`. | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES` | int / `16384` | Per `raw auth key + session_id` owner/receipt cap. | | `TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH` | int / `2048` | Additional active-owner cap per raw auth key; no greater than global pending tasks or auth entries. | | `TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | Process-wide reservation for transport wire bytes, maximum decrypted plaintext, and every live outer/nested gzip expansion, acquired before the corresponding payload allocation. | | `TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE` | int / `128` | Per-connection normal outbound mailbox capacity. | | `TELESRV_MTPROTO_OUTBOUND_CONTROL_QUEUE_SIZE` | int / `32` | Per-connection control-message mailbox capacity. | -| `TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | Global budget for tracked resend-pending message bodies. | +| `TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | Sole global budget for unacknowledged logical-session bodies. Reconnects reuse the same `msg_id/seq_no/body`; ACK, destroy, or six minutes offline releases it, with no second RPC cache/spool copy. | | `TELESRV_MTPROTO_OUTBOUND_WRITE_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | Global budget for concurrent encrypted wire/codec/obfuscation scratch. | Nested gzip admission adds no environment setting. Code-enforced ceilings are diff --git a/docs/configuration.zh-CN.md b/docs/configuration.zh-CN.md index 117da710..acdf9b21 100644 --- a/docs/configuration.zh-CN.md +++ b/docs/configuration.zh-CN.md @@ -35,17 +35,14 @@ | `TELESRV_MTPROTO_RPC_GLOBAL_WORKERS` | int / `256` | 共享公平调度器 worker 数。 | | `TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS` | int / `8192` | 进程级排队与执行中的 RPC task 上限。 | | `TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES` | int64 charge bytes / `536870912` | 进程级已预留/排队/执行中 RPC 内存 charge 预算；legacy 等于 copied body，exact 是 typed decode 前按 wire 与生成对象放大计算的保守 materialization charge。nested gzip 展开后会在 decoder 分配 typed graph 前原子增长该 charge，grow 失败原子拒绝整批候选 RPC；该值不代表可并发接收同等大小的 wire body。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES` | int / `262144` | 331 秒进程内重放窗口中，pending owner、completed `rpc_result` 与容量 tombstone 的全局 ownership 条目上限。owner 执行前先占 1 条，转 completed 时不重复计数。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_BYTES` | int64 bytes / `67108864` | 上述 ownership 的全局 retained-byte 上限；owner 先占 1 byte，Put 转移为真实 body 或 1-byte identity tombstone。不得低于 `16775168`（单条合法 outbound body 上限）。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES` | int / `32768` | 单 raw auth key 的 ownership 条目上限；与全局、session 层同时计费，防一个 auth key 吃满进程缓存。必须 `global >= auth >= session`。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES` | int64 bytes / `33554432` | 单 raw auth key retained-byte 上限；必须不低于单条合法 outbound body，且满足 byte 层级关系。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES` | int / `16384` | 单 `raw auth key + session_id` ownership 条目上限；不同 session 不共享该局部额度。 | -| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES` | int64 bytes / `16777216` | 单 `raw auth key + session_id` retained-byte 上限；默认略高于单条合法 outbound body，确保空预算时任一合法结果可完整进入。 | -| `TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH` | int / `2048` | 单 raw auth key 的 active pending owner 附加上限；必须不大于 `RPC_GLOBAL_MAX_TASKS` 和 auth entry 上限。Put/Abort 都立即归还此 active 额度。 | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES` | int / `262144` | pending owner 与未 ACK 的轻量结果收据全局上限。收据只存请求身份、执行结果和 Layer admission 元数据，不存 TL body；收到 `msgs_ack` 立即删除，331 秒仅是无 ACK 时的安全上限。 | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES` | int / `32768` | 单 raw auth key 的 owner/收据条目上限；必须满足 `global >= auth >= session`。 | +| `TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES` | int / `16384` | 单 `raw auth key + session_id` 的 owner/收据条目上限。 | +| `TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH` | int / `2048` | 单 raw auth key 的 active pending owner 附加上限；必须不大于 `RPC_GLOBAL_MAX_TASKS` 和 auth entry 上限。 | | `TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | transport wire、最大解密明文以及每个 live outer/nested gzip 输出的进程级在途预算，均在对应 payload 分配前预留。 | | `TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE` | int / `128` | 单连接普通 outbound mailbox 容量。 | | `TELESRV_MTPROTO_OUTBOUND_CONTROL_QUEUE_SIZE` | int / `32` | 单连接控制消息 mailbox 容量。 | -| `TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | resend pending message body 的全局预算。 | +| `TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | 所有逻辑 session 未 ACK 出站 body 的唯一全局预算。物理连接重连复用同一份 `msg_id/seq_no/body`；ACK、destroy 或离线 6 分钟回收时释放，不再另建 RPC cache/spool 副本。 | | `TELESRV_MTPROTO_OUTBOUND_WRITE_GLOBAL_MAX_BYTES` | int64 bytes / `536870912` | 并发加密 wire/codec/obfuscation scratch 的全局预算。 | nested gzip admission 不新增环境变量。代码硬限制为：每个 diff --git a/internal/config/config.go b/internal/config/config.go index 0702a2d8..821418b9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -60,14 +60,12 @@ type Config struct { MTProtoRPCGlobalWorkers int MTProtoRPCGlobalMaxTasks int MTProtoRPCGlobalMaxBytes int64 - // Pending ownership and completed rpc_result replay state share a three-level - // global/raw-auth/session budget over the full MTProto duplicate horizon. + // Pending ownership and compact completed receipts share three-level + // global/raw-auth/session entry accounting. Result bodies are never cached + // here; the logical-session outbox owns unacknowledged wire bytes. MTProtoRPCResultCacheMaxEntries int - MTProtoRPCResultCacheMaxBytes int64 MTProtoRPCResultCacheAuthMaxEntries int - MTProtoRPCResultCacheAuthMaxBytes int64 MTProtoRPCResultCacheSessionMaxEntries int - MTProtoRPCResultCacheSessionMaxBytes int64 MTProtoRPCResultPendingPerAuth int // MTProtoInboundFrameGlobalMaxBytes 是 transport wire + 最大解密 plaintext 的 // 进程级在途预算；frame 长度读出后、payload 分配前预留。 @@ -571,6 +569,9 @@ func Load() (Config, error) { envInt64Or := fileEnv.envInt64Or envDurationOr := fileEnv.envDurationOr envAllowEmptyOr := fileEnv.envAllowEmptyOr + if err := validateStrictMTProtoCapacityEnv(fileEnv); err != nil { + return Config{}, err + } publicBaseURL, err := links.ValidateBaseURL(envOr("TELESRV_PUBLIC_BASE_URL", links.DefaultPublicBaseURL)) if err != nil { @@ -629,15 +630,10 @@ func Load() (Config, error) { MTProtoRPCGlobalMaxTasks: envIntOr("TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS", 8192), MTProtoRPCGlobalMaxBytes: envInt64Or("TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES", 512<<20), MTProtoRPCResultCacheMaxEntries: envIntOr("TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES", 1<<18), - MTProtoRPCResultCacheMaxBytes: envInt64Or("TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_BYTES", 64<<20), MTProtoRPCResultCacheAuthMaxEntries: envIntOr("TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES", 1<<15), - MTProtoRPCResultCacheAuthMaxBytes: envInt64Or("TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES", 32<<20), MTProtoRPCResultCacheSessionMaxEntries: envIntOr( "TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES", 1<<14, ), - MTProtoRPCResultCacheSessionMaxBytes: envInt64Or( - "TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES", 16<<20, - ), MTProtoRPCResultPendingPerAuth: envIntOr("TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH", 1<<11), MTProtoInboundFrameGlobalMaxBytes: envInt64Or("TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES", 512<<20), MTProtoOutboundQueueSize: envIntOr("TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE", 128), @@ -1211,8 +1207,6 @@ func validateCollectibleUsernameConfig(cfg Config) error { return nil } -const mtProtoRPCResultMinBytes = int64((1 << 24) - (2 << 10)) - func validateRPCResultCacheConfig(cfg Config) error { if cfg.MTProtoRPCResultCacheMaxEntries <= 0 || cfg.MTProtoRPCResultCacheAuthMaxEntries <= 0 || cfg.MTProtoRPCResultCacheSessionMaxEntries <= 0 { @@ -1223,18 +1217,6 @@ func validateRPCResultCacheConfig(cfg Config) error { return fmt.Errorf("MTProto rpc_result entry hierarchy must satisfy global >= auth >= session: %d/%d/%d", cfg.MTProtoRPCResultCacheMaxEntries, cfg.MTProtoRPCResultCacheAuthMaxEntries, cfg.MTProtoRPCResultCacheSessionMaxEntries) } - if cfg.MTProtoRPCResultCacheMaxBytes < mtProtoRPCResultMinBytes || - cfg.MTProtoRPCResultCacheAuthMaxBytes < mtProtoRPCResultMinBytes || - cfg.MTProtoRPCResultCacheSessionMaxBytes < mtProtoRPCResultMinBytes { - return fmt.Errorf("MTProto rpc_result byte limits must each be at least %d: %d/%d/%d", - mtProtoRPCResultMinBytes, cfg.MTProtoRPCResultCacheMaxBytes, - cfg.MTProtoRPCResultCacheAuthMaxBytes, cfg.MTProtoRPCResultCacheSessionMaxBytes) - } - if cfg.MTProtoRPCResultCacheMaxBytes < cfg.MTProtoRPCResultCacheAuthMaxBytes || - cfg.MTProtoRPCResultCacheAuthMaxBytes < cfg.MTProtoRPCResultCacheSessionMaxBytes { - return fmt.Errorf("MTProto rpc_result byte hierarchy must satisfy global >= auth >= session: %d/%d/%d", - cfg.MTProtoRPCResultCacheMaxBytes, cfg.MTProtoRPCResultCacheAuthMaxBytes, cfg.MTProtoRPCResultCacheSessionMaxBytes) - } if cfg.MTProtoRPCGlobalMaxTasks <= 0 || cfg.MTProtoRPCResultPendingPerAuth <= 0 || cfg.MTProtoRPCResultPendingPerAuth > cfg.MTProtoRPCGlobalMaxTasks || cfg.MTProtoRPCResultPendingPerAuth > cfg.MTProtoRPCResultCacheAuthMaxEntries { @@ -1525,6 +1507,43 @@ func (e envSource) envIntOr(key string, def int) int { return def } +func validateStrictMTProtoCapacityEnv(e envSource) error { + for _, key := range []string{ + "TELESRV_MTPROTO_MAX_CONNECTIONS", + "TELESRV_MTPROTO_MAX_CONNECTIONS_PER_IP", + "TELESRV_MTPROTO_MAX_CONCURRENT_HANDSHAKES", + "TELESRV_MTPROTO_RPC_MAX_INFLIGHT", + "TELESRV_MTPROTO_RPC_QUEUE_SIZE", + "TELESRV_MTPROTO_RPC_GLOBAL_WORKERS", + "TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS", + "TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES", + "TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES", + "TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES", + "TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH", + "TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE", + "TELESRV_MTPROTO_OUTBOUND_CONTROL_QUEUE_SIZE", + } { + if raw := e.envOr(key, ""); raw != "" { + if _, err := strconv.Atoi(raw); err != nil { + return fmt.Errorf("%s must be a base-10 integer: %w", key, err) + } + } + } + for _, key := range []string{ + "TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES", + "TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES", + "TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES", + "TELESRV_MTPROTO_OUTBOUND_WRITE_GLOBAL_MAX_BYTES", + } { + if raw := e.envOr(key, ""); raw != "" { + if _, err := strconv.ParseInt(raw, 10, 64); err != nil { + return fmt.Errorf("%s must be a base-10 int64: %w", key, err) + } + } + } + return nil +} + func (e envSource) envInt64Or(key string, def int64) int64 { if v := e.envOr(key, ""); v != "" { if n, err := strconv.ParseInt(v, 10, 64); err == nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 0ccb0b40..7f5c619c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -151,11 +151,8 @@ func TestLoadMTProtoAdmissionAndRPCBudgets(t *testing.T) { t.Setenv("TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS", "444") t.Setenv("TELESRV_MTPROTO_RPC_GLOBAL_MAX_BYTES", "555555") t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES", "555") - t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_BYTES", "70000000") t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_ENTRIES", "444") - t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES", "40000000") t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_ENTRIES", "333") - t.Setenv("TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES", "20000000") t.Setenv("TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH", "222") t.Setenv("TELESRV_MTPROTO_INBOUND_FRAME_GLOBAL_MAX_BYTES", "777777") t.Setenv("TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE", "88") @@ -177,14 +174,14 @@ func TestLoadMTProtoAdmissionAndRPCBudgets(t *testing.T) { cfg.MTProtoRPCGlobalWorkers != 33 || cfg.MTProtoRPCGlobalMaxTasks != 444 || cfg.MTProtoRPCGlobalMaxBytes != 555555 { t.Fatalf("rpc budget config = %d/%d/%v/%d/%d/%d", cfg.MTProtoRPCMaxInflight, cfg.MTProtoRPCQueueSize, cfg.MTProtoRPCTimeout, cfg.MTProtoRPCGlobalWorkers, cfg.MTProtoRPCGlobalMaxTasks, cfg.MTProtoRPCGlobalMaxBytes) } - if cfg.MTProtoRPCResultCacheMaxEntries != 555 || cfg.MTProtoRPCResultCacheMaxBytes != 70000000 || - cfg.MTProtoRPCResultCacheAuthMaxEntries != 444 || cfg.MTProtoRPCResultCacheAuthMaxBytes != 40000000 || - cfg.MTProtoRPCResultCacheSessionMaxEntries != 333 || cfg.MTProtoRPCResultCacheSessionMaxBytes != 20000000 || + if cfg.MTProtoRPCResultCacheMaxEntries != 555 || + cfg.MTProtoRPCResultCacheAuthMaxEntries != 444 || + cfg.MTProtoRPCResultCacheSessionMaxEntries != 333 || cfg.MTProtoRPCResultPendingPerAuth != 222 { - t.Fatalf("rpc result cache config = global:%d/%d auth:%d/%d session:%d/%d pending/auth:%d", - cfg.MTProtoRPCResultCacheMaxEntries, cfg.MTProtoRPCResultCacheMaxBytes, - cfg.MTProtoRPCResultCacheAuthMaxEntries, cfg.MTProtoRPCResultCacheAuthMaxBytes, - cfg.MTProtoRPCResultCacheSessionMaxEntries, cfg.MTProtoRPCResultCacheSessionMaxBytes, + t.Fatalf("rpc result receipt config = global:%d auth:%d session:%d pending/auth:%d", + cfg.MTProtoRPCResultCacheMaxEntries, + cfg.MTProtoRPCResultCacheAuthMaxEntries, + cfg.MTProtoRPCResultCacheSessionMaxEntries, cfg.MTProtoRPCResultPendingPerAuth) } if cfg.MTProtoInboundFrameGlobalMaxBytes != 777777 { @@ -204,14 +201,14 @@ func TestLoadRPCResultFairBudgetDefaults(t *testing.T) { if err != nil { t.Fatalf("Load: %v", err) } - if cfg.MTProtoRPCResultCacheMaxEntries != 1<<18 || cfg.MTProtoRPCResultCacheMaxBytes != 64<<20 || - cfg.MTProtoRPCResultCacheAuthMaxEntries != 1<<15 || cfg.MTProtoRPCResultCacheAuthMaxBytes != 32<<20 || - cfg.MTProtoRPCResultCacheSessionMaxEntries != 1<<14 || cfg.MTProtoRPCResultCacheSessionMaxBytes != 16<<20 || + if cfg.MTProtoRPCResultCacheMaxEntries != 1<<18 || + cfg.MTProtoRPCResultCacheAuthMaxEntries != 1<<15 || + cfg.MTProtoRPCResultCacheSessionMaxEntries != 1<<14 || cfg.MTProtoRPCResultPendingPerAuth != 1<<11 { - t.Fatalf("rpc_result fair defaults = global:%d/%d auth:%d/%d session:%d/%d pending/auth:%d", - cfg.MTProtoRPCResultCacheMaxEntries, cfg.MTProtoRPCResultCacheMaxBytes, - cfg.MTProtoRPCResultCacheAuthMaxEntries, cfg.MTProtoRPCResultCacheAuthMaxBytes, - cfg.MTProtoRPCResultCacheSessionMaxEntries, cfg.MTProtoRPCResultCacheSessionMaxBytes, + t.Fatalf("rpc_result receipt defaults = global:%d auth:%d session:%d pending/auth:%d", + cfg.MTProtoRPCResultCacheMaxEntries, + cfg.MTProtoRPCResultCacheAuthMaxEntries, + cfg.MTProtoRPCResultCacheSessionMaxEntries, cfg.MTProtoRPCResultPendingPerAuth) } } @@ -223,8 +220,6 @@ func TestLoadRejectsInvalidRPCResultFairBudgets(t *testing.T) { value string }{ {name: "entry hierarchy", key: "TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES", value: "1024"}, - {name: "byte below outbound body", key: "TELESRV_MTPROTO_RPC_RESULT_CACHE_SESSION_MAX_BYTES", value: "16700000"}, - {name: "byte hierarchy", key: "TELESRV_MTPROTO_RPC_RESULT_CACHE_AUTH_MAX_BYTES", value: "70000000"}, {name: "pending hierarchy", key: "TELESRV_MTPROTO_RPC_RESULT_PENDING_PER_AUTH", value: "9000"}, } for _, test := range tests { @@ -238,6 +233,27 @@ func TestLoadRejectsInvalidRPCResultFairBudgets(t *testing.T) { } } +func TestLoadRejectsMalformedMTProtoCapacity(t *testing.T) { + for _, test := range []struct { + name string + key string + value string + }{ + {name: "worker tasks malformed", key: "TELESRV_MTPROTO_RPC_GLOBAL_MAX_TASKS", value: "lots"}, + {name: "receipt entries overflow", key: "TELESRV_MTPROTO_RPC_RESULT_CACHE_MAX_ENTRIES", value: "999999999999999999999999"}, + {name: "tracked bytes overflow", key: "TELESRV_MTPROTO_OUTBOUND_TRACKED_GLOBAL_MAX_BYTES", value: "999999999999999999999999"}, + {name: "outbound queue malformed", key: "TELESRV_MTPROTO_OUTBOUND_QUEUE_SIZE", value: "many"}, + } { + t.Run(test.name, func(t *testing.T) { + disableDefaultConfigFile(t) + t.Setenv(test.key, test.value) + if _, err := Load(); err == nil { + t.Fatalf("Load accepted %s=%q", test.key, test.value) + } + }) + } +} + func TestLoadOutboxPoisonPolicy(t *testing.T) { disableDefaultConfigFile(t) t.Setenv("TELESRV_OUTBOX_POISON_RETENTION", "2m") diff --git a/internal/mtprotoedge/conn.go b/internal/mtprotoedge/conn.go index 2ae0d545..447d79d1 100644 --- a/internal/mtprotoedge/conn.go +++ b/internal/mtprotoedge/conn.go @@ -108,6 +108,9 @@ type Conn struct { outboundControlBudgetOnce sync.Once outboundScratchPool *outboundScratchPool outboundScratchOnce sync.Once + // outboundState outlives this physical Conn generation. A replacement + // physical connection for the same auth key/session reuses it. + outboundState *outboundState // lifecycle is the sole monotonic activation/retirement state machine. // retired never transitions back to claiming/active; one atomic state avoids // contradictory activation and shutdown observations. @@ -150,7 +153,8 @@ type Conn struct { rpcRootCtx context.Context rpcMaxInflight int - // sentContentMessages 只由 outbound actor 访问，用于生成 MTProto seq_no。 + // sentContentMessages is retained only for standalone construction tests. + // Server connections allocate seq_no from logical-session outboundState. sentContentMessages int32 // outboundRand 只由 outbound actor 访问：对 cipher 随机源的缓冲预读， // 把每帧 padding 的 getrandom syscall 摊薄成 ~1KiB 一次。 diff --git a/internal/mtprotoedge/encrypted.go b/internal/mtprotoedge/encrypted.go index 96cc0f23..20eb29c7 100644 --- a/internal/mtprotoedge/encrypted.go +++ b/internal/mtprotoedge/encrypted.go @@ -716,10 +716,7 @@ func (s *Server) handleInboundRPCAdmissionError(ctx context.Context, c *Conn, ms zap.String("auth_key_id", c.authKeyHex), zap.Int64("session_id", c.sessionID), ) - return s.sendResult(ctx, c, msgID, &mt.RPCError{ - ErrorCode: 420, - ErrorMessage: "FLOOD_WAIT_1", - }) + return s.sendResult(ctx, c, msgID, rpcWorkerBusyError()) } return err } @@ -853,9 +850,9 @@ var errRPCResultRetentionHandoff = errors.New("mtproto rpc result retention hand type rpcResultRetentionHandoff func(*encodedOutboundMessage, error) error // publishRPCResult ends the inbound worker's ownership at bounded egress -// admission. Physical delivery is thereafter owned either by the single -// outbound actor or, under retained-byte saturation, by a fenced completed-cache -// entry that the replacement connection can replay without rerunning business. +// admission. Physical delivery is thereafter owned by the logical-session +// outbox. Under retained-byte saturation the Conn is fenced and the receipt +// ledger records an unavailable tombstone so business cannot rerun. func (s *Server) publishRPCResult( c *Conn, reqMsgID int64, @@ -893,21 +890,19 @@ func (s *Server) publishRPCResult( return priority, visible } - // A successful business result may never leave the encode slot as an - // unaccounted []byte. If the primary 512MiB retained-body budget is full, make - // overload terminal for this physical generation and publish the exact result - // into the independently bounded completed cache before releasing the slot. + // If the sole logical-outbox body budget cannot admit a completed result, + // fence this physical generation and publish only an execution tombstone. + // There is deliberately no fallback payload cache/spool and no business + // re-execution hidden behind a local capacity error. retainForReplay := func(encoded *encodedOutboundMessage, admissionErr error) error { if s == nil || s.rpcResults == nil || c == nil || encoded == nil || reqMsgID == 0 { - return errors.New("rpc result completed cache is unavailable") + return errors.New("rpc result receipt ledger is unavailable") } - if int64(len(encoded.body)) > s.rpcResults.completedBytes.max { - // Every transport-legal result fits the production completed cache by the - // compile-time invariant in rpc_result_cache.go. A test/custom cache that - // violates it cannot safely complete this flight, so fail fast while the - // body is still confined to the encode slot. + if s.rpcResults.sessions == nil && int64(len(encoded.body)) > s.rpcResults.completedBytes.max { + // Focused legacy-cache tests may intentionally use a byte budget smaller + // than one legal transport result. Production receipts never own bodies. panic(fmt.Sprintf( - "mtprotoedge: encoded rpc result exceeds completed-cache budget: body=%d max=%d", + "mtprotoedge: legacy encoded rpc result exceeds inline-ledger budget: body=%d max=%d", len(encoded.body), s.rpcResults.completedBytes.max, )) } @@ -929,7 +924,7 @@ func (s *Server) publishRPCResult( if visible { resultLogLevel = zap.InfoLevel } - if checked := s.log.Check(resultLogLevel, "RPC result retained for replay after egress saturation"); checked != nil { + if checked := s.log.Check(resultLogLevel, "RPC result execution fenced after egress saturation"); checked != nil { checked.Write( zap.String("method", method), zap.Int64("req_msg_id", reqMsgID), zap.Int64("delivered_req_msg_id", encoded.writtenRequestID()), @@ -1070,7 +1065,7 @@ func (s *Server) sendResult(ctx context.Context, c *Conn, reqMsgID int64, result } // sendCachedRPCResult preserves the delivery half of the rpc_result invariant -// for completed-flight replays: either the cached result reaches this physical +// for completed-flight replays: either the logical outbox result reaches this physical // byte stream, or this logical Conn is fenced so a replacement may retry it. func (s *Server) sendCachedRPCResult(ctx context.Context, c *Conn, encoded *encodedOutboundMessage) error { return s.sendCachedRPCResultWithHook(ctx, c, encoded, nil) @@ -1251,8 +1246,9 @@ func (s *Server) encodeRPCResultReservedWithHandoffContext( } retained = true // The handoff owns the only surviving pointer. Do not return a second - // producer reference after the encode slot releases; the completed cache - // may independently evict the entry under its bounded policy. + // producer reference after the encode slot releases. Production handoff + // either transferred the body to the logical outbox or retained only an + // unavailable receipt tombstone. encoded = nil return admissionErr } @@ -1355,6 +1351,7 @@ func (s *Server) storeRPCResult(c *Conn, reqMsgID int64, encoded *encodedOutboun if s == nil || s.rpcResults == nil || c == nil { return } + s.conns.adoptLogicalSession(c) s.rpcResults.Put(c.authKeyID, c.sessionID, reqMsgID, encoded) } diff --git a/internal/mtprotoedge/inbound_layer_rpc.go b/internal/mtprotoedge/inbound_layer_rpc.go index 5c419372..c32adee0 100644 --- a/internal/mtprotoedge/inbound_layer_rpc.go +++ b/internal/mtprotoedge/inbound_layer_rpc.go @@ -434,6 +434,10 @@ func (s *Server) prepareInboundLayerRPCBatch(ctx context.Context, c *Conn, plan } switch claim.state { case rpcResultAcquireCompleted: + // Transfer the materialization ticket to the plan before any + // profile-restore step can fail; plan.close is the universal abort + // path and sendCached releases it after outbound-budget handoff. + item.payload = claim.encoded after, prepareErr := s.prepareAdmittedLayerRPCReplay(ctx, c, item.msgID, claim.admissionSeq, item.profileEvidenceFresh(), item.admitted) if prepareErr != nil { s.rpcRewrap.release(candidate) @@ -441,10 +445,16 @@ func (s *Server) prepareInboundLayerRPCBatch(ctx context.Context, c *Conn, plan } s.rpcRewrap.commit(candidate) item.kind = inboundItemReplayRPC - item.payload = claim.encoded if claim.executionKnown && claim.executionOK { item.replayAfterSuccessfulDelivery = after } + case rpcResultAcquireAcknowledged: + // Explicit ACK is terminal proof for this request msg_id. Keep + // exact admission metadata, retire the old rewrap candidate and + // make the duplicate ACK-only without replay side effects. + s.rpcRewrap.commit(candidate) + item.kind = inboundItemDuplicate + item.payload = nil case rpcResultAcquirePending: after, prepareErr := s.prepareAdmittedLayerRPCReplay(ctx, c, item.msgID, claim.admissionSeq, item.profileEvidenceFresh(), item.admitted) if prepareErr != nil { @@ -523,15 +533,18 @@ func (s *Server) prepareInboundLayerRPCBatch(ctx context.Context, c *Conn, plan } switch claim.state { case rpcResultAcquireCompleted: + item.payload = claim.encoded after, prepareErr := s.prepareAdmittedLayerRPCReplay(ctx, c, item.msgID, claim.admissionSeq, item.profileEvidenceFresh(), item.admitted) if prepareErr != nil { return prepareErr } item.kind = inboundItemReplayRPC - item.payload = claim.encoded if claim.executionKnown && claim.executionOK { item.replayAfterSuccessfulDelivery = after } + case rpcResultAcquireAcknowledged: + item.kind = inboundItemDuplicate + item.payload = nil case rpcResultAcquirePending: if ownersInPlan[item.msgID] != nil { item.kind = inboundItemDuplicate diff --git a/internal/mtprotoedge/inbound_layer_rpc_test.go b/internal/mtprotoedge/inbound_layer_rpc_test.go index 90905557..313c2206 100644 --- a/internal/mtprotoedge/inbound_layer_rpc_test.go +++ b/internal/mtprotoedge/inbound_layer_rpc_test.go @@ -541,7 +541,7 @@ func TestBatchProvisionalCursorKeepsRegistryWatermarkAcrossOldReplay(t *testing. t.Fatalf("old owner err=%v", err) } oldClaim.owner.CompleteExecution(true) - s.rpcResults.Put(authKeyID, sessionID, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) + storeLogicalRPCResultForTest(t, s, c, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) nakedBody := exactOutboundLayerRPCBody(t, tlprofile.Profile227, &tg.MessagesGetHistoryRequest{ Peer: &tg.InputPeerSelf{}, Limit: 1, @@ -562,6 +562,131 @@ func TestBatchProvisionalCursorKeepsRegistryWatermarkAcrossOldReplay(t *testing. } } +func TestAcknowledgedExactRPCReleasesReplayReceipt(t *testing.T) { + handler := &replayProfileCaptureLayerRPC{admissionOnlyLayerRPC: newAdmissionOnlyLayerRPC()} + s := New(Options{DC: 2, LayerRPC: handler}) + authKeyID := [8]byte{0x31, 0xa1} + const sessionID, reqMsgID = int64(3191), int64(100) + body := exactLayerRPCBody(t, &tg.InvokeWithLayerRequest{ + Layer: 225, Query: &tg.HelpGetConfigRequest{}, + }) + admitted, _, err := s.decodeInboundLayerRPC( + LayerProfileSnapshot{Profile: tlprofile.Profile225, Origin: LayerProfileExplicit}, body, + ) + if err != nil { + t.Fatal(err) + } + owner, err := s.rpcResults.AcquireLayerIdentified( + authKeyID, sessionID, reqMsgID, + admitted.Call().Profile(), admitted.Prepared().Identity(), + ) + if err != nil || owner.state != rpcResultAcquireOwner { + t.Fatalf("owner = %#v, err=%v", owner, err) + } + owner.owner.CompleteExecution(true) + logical := &Conn{authKeyID: authKeyID, sessionID: sessionID} + storeLogicalRPCResultForTest(t, s, logical, reqMsgID, &encodedOutboundMessage{ + body: make([]byte, 32), reqMsgID: reqMsgID, + }) + acknowledgeLogicalRPCResultForTest(t, s, logical, reqMsgID) + + replacement := &Conn{authKeyID: authKeyID, sessionID: sessionID, metrics: NopMetrics{}} + replacement.startInboundRPCScheduler(s.rpcScheduler, 1, 8, time.Second) + defer replacement.Close() + plan := &inboundPlan{items: []inboundItem{{kind: inboundItemRPC, msgID: reqMsgID, body: body}}} + defer plan.close() + if err := s.prepareInboundLayerRPCBatch(context.Background(), replacement, plan); err != nil { + t.Fatal(err) + } + if plan.items[0].kind != inboundItemRPC || plan.items[0].payload == nil { + t.Fatalf("post-ACK request = kind:%d payload:%T", plan.items[0].kind, plan.items[0].payload) + } + if len(plan.rpcTasks) != 1 || len(plan.rpcOwners) != 1 || len(plan.rewrapAliases) != 0 || plan.rpcReservation == nil { + t.Fatalf("post-ACK request scheduling: tasks=%d owners=%d aliases=%d reservation=%v", + len(plan.rpcTasks), len(plan.rpcOwners), len(plan.rewrapAliases), plan.rpcReservation != nil) + } + if profiles, known := handler.capturedProfiles(); len(profiles) != 0 || len(known) != 0 { + t.Fatalf("ACKed duplicate ran replay side effects: profiles=%v known=%v", profiles, known) + } + claim, err := s.rpcResults.AcquireLayerIdentified( + authKeyID, sessionID, reqMsgID, + admitted.Call().Profile(), admitted.Prepared().Identity(), + ) + if err != nil || claim.state != rpcResultAcquirePending || claim.admissionSeq == owner.admissionSeq { + t.Fatalf("post-ACK claim after preflight = %#v, err=%v", claim, err) + } +} + +func TestAcknowledgedExactInitRewrapReleasesReplayReceipt(t *testing.T) { + handler := &replayProfileCaptureLayerRPC{admissionOnlyLayerRPC: newAdmissionOnlyLayerRPC()} + s := New(Options{DC: 2, LayerRPC: handler}) + authKeyID := [8]byte{0x31, 0xa2} + const sessionID, oldReqID, newReqID = int64(3192), int64(100), int64(104) + c := &Conn{authKeyID: authKeyID, sessionID: sessionID, metrics: NopMetrics{}} + if err := c.SeedInheritedLayerProfile(tlprofile.Profile227); err != nil { + t.Fatal(err) + } + c.startInboundRPCScheduler(s.rpcScheduler, 1, 8, time.Second) + defer c.Close() + + inner := exactOutboundLayerRPCBody(t, tlprofile.Profile227, &tg.HelpGetConfigRequest{}) + oldPlan := &inboundPlan{items: []inboundItem{{kind: inboundItemRPC, msgID: oldReqID, body: inner}}} + defer oldPlan.close() + if err := s.prepareInboundLayerRPCBatch(context.Background(), c, oldPlan); err != nil { + t.Fatal(err) + } + if len(oldPlan.rpcOwners) != 1 || s.rpcRewrap.total != 1 { + t.Fatalf("old exact candidate = owners:%d candidates:%d", len(oldPlan.rpcOwners), s.rpcRewrap.total) + } + + wrapped := exactLayerRPCBody(t, &tg.InvokeWithLayerRequest{ + Layer: 227, + Query: &tg.InitConnectionRequest{ + APIID: 6, DeviceModel: "Pixel", SystemVersion: "SDK 36", AppVersion: "12.8.1", + SystemLangCode: "en", LangPack: "android", LangCode: "en", + Query: &tg.HelpGetConfigRequest{}, + }, + }) + admitted, _, err := s.decodeInboundLayerRPC( + LayerProfileSnapshot{Profile: tlprofile.Profile227, Origin: LayerProfileExplicit}, wrapped, + ) + if err != nil { + t.Fatal(err) + } + newOwner, err := s.rpcResults.AcquireLayerIdentified( + authKeyID, sessionID, newReqID, + admitted.Call().Profile(), admitted.Prepared().Identity(), + ) + if err != nil || newOwner.state != rpcResultAcquireOwner { + t.Fatalf("new exact receipt owner = %#v, err=%v", newOwner, err) + } + newOwner.owner.CompleteExecution(true) + storeLogicalRPCResultForTest(t, s, c, newReqID, &encodedOutboundMessage{ + body: make([]byte, 32), reqMsgID: newReqID, + }) + acknowledgeLogicalRPCResultForTest(t, s, c, newReqID) + // In the real delivery path a successful rewrap commits its source candidate. + // Clear the synthetic pending candidate before observing post-ACK admission. + s.rpcRewrap.clearSession(c) + + newPlan := &inboundPlan{items: []inboundItem{{kind: inboundItemRPC, msgID: newReqID, body: wrapped}}} + defer newPlan.close() + if err := s.prepareInboundLayerRPCBatch(context.Background(), c, newPlan); err != nil { + t.Fatal(err) + } + if newPlan.items[0].kind != inboundItemRPC || newPlan.items[0].payload == nil || + len(newPlan.rpcTasks) != 1 || len(newPlan.rpcOwners) != 1 || len(newPlan.rewrapAliases) != 0 { + t.Fatalf("post-ACK init request scheduling: kind=%d tasks=%d owners=%d aliases=%d", + newPlan.items[0].kind, len(newPlan.rpcTasks), len(newPlan.rpcOwners), len(newPlan.rewrapAliases)) + } + if s.rpcRewrap.total != 0 { + t.Fatalf("ACKed exact init rewrap left %d candidates", s.rpcRewrap.total) + } + if profiles, known := handler.capturedProfiles(); len(profiles) != 0 || len(known) != 0 { + t.Fatalf("ACKed exact init rewrap ran replay side effects: profiles=%v known=%v", profiles, known) + } +} + func TestBatchProvisionalCursorUsesPendingNewerExplicitEvidence(t *testing.T) { handler := newAdmissionOnlyLayerRPC() s := New(Options{DC: 2, LayerRPC: handler}) @@ -921,7 +1046,7 @@ func TestInvariantReplayNeverCachesInternalCanonicalProfile(t *testing.T) { } owner.CompleteExecution(true) - s.rpcResults.Put(authKeyID, sessionID, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) + storeLogicalRPCResultForTest(t, s, original, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) if profile, ok := s.rpcResults.ExactAdmissionProfile(authKeyID, sessionID, 100); ok || profile != 0 { t.Fatalf("completed invariant cached profile=(%d,%v)", profile, ok) } @@ -1121,7 +1246,7 @@ func TestOldCompletedLayerRequestCannotRollBackCorrectedSession(t *testing.T) { t.Fatal("old request did not acquire owner") } oldOwner.CompleteExecution(true) - s.rpcResults.Put(authKeyID, sessionID, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) + storeLogicalRPCResultForTest(t, s, c, 100, &encodedOutboundMessage{body: []byte{1}, reqMsgID: 100}) correctPlan := &inboundPlan{items: []inboundItem{{ kind: inboundItemRPC, msgID: 104, @@ -1187,7 +1312,7 @@ func TestLogicalSessionLayerWatermarkSurvivesResultExpiryAndOldContainer(t *test if oldOwner == nil || !oldOwner.CompleteExecution(true) { t.Fatal("old Layer 225 request did not establish a completed owner") } - s.rpcResults.Put(authKeyID, sessionID, oldMsgID, &encodedOutboundMessage{body: []byte{1}, reqMsgID: oldMsgID}) + storeLogicalRPCResultForTest(t, s, original, oldMsgID, &encodedOutboundMessage{body: []byte{1}, reqMsgID: oldMsgID}) correctPlan := &inboundPlan{items: []inboundItem{{ kind: inboundItemRPC, msgID: correctedMsgID, diff --git a/internal/mtprotoedge/inbound_preflight.go b/internal/mtprotoedge/inbound_preflight.go index 4755041b..607db7ee 100644 --- a/internal/mtprotoedge/inbound_preflight.go +++ b/internal/mtprotoedge/inbound_preflight.go @@ -732,8 +732,8 @@ func preflightInboundItem(msgID int64, seqNo int32, typeID uint32, content bool, // prepareInboundRPCBatch performs the whole container's count/byte admission // before copying or scheduling any API RPC. Capacity exhaustion is converted -// into one consistent terminal FLOOD_WAIT result per uncached RPC; no business -// handler from the batch is allowed to start in that case. +// into one consistent local 500 WORKER_BUSY_TOO_LONG_RETRY result per uncached +// RPC; no business handler from the batch is allowed to start in that case. func (s *Server) prepareInboundRPCBatch(ctx context.Context, c *Conn, plan *inboundPlan) error { if s.layerRPC != nil { return s.prepareInboundLayerRPCBatch(ctx, c, plan) @@ -784,6 +784,13 @@ func (s *Server) prepareInboundRPCBatch(ctx context.Context, c *Conn, plan *inbo s.rpcRewrap.commit(candidate) item.kind = inboundItemReplayRPC item.payload = claim.encoded + case rpcResultAcquireAcknowledged: + // The client already ACKed the correlated rpc_result. Retire the + // stale rewrap candidate and keep this duplicate ACK-only; neither + // the old body nor the business handler may run again. + s.rpcRewrap.commit(candidate) + item.kind = inboundItemDuplicate + item.payload = nil case rpcResultAcquirePending: s.rpcRewrap.commit(candidate) item.kind = inboundItemRewrappedRPC @@ -847,6 +854,9 @@ func (s *Server) prepareInboundRPCBatch(ctx context.Context, c *Conn, plan *inbo ) item.kind = inboundItemReplayRPC item.payload = claim.encoded + case rpcResultAcquireAcknowledged: + item.kind = inboundItemDuplicate + item.payload = nil case rpcResultAcquirePending: // A malformed/replayed container may repeat the same msg_id after this // very plan installed its owner. More generally, any request already in @@ -1026,6 +1036,10 @@ func (s *Server) executeInboundPlan(ctx context.Context, cs *connState, c *Conn, }); err != nil { return err } + // The key cannot reconnect to ACK or replay any old answer. Release every + // logical outbox and receipt only after the terminal OK is physically on + // the wire; retaining them for the offline TTL would be pure leakage. + s.conns.ForgetLogicalSessionsForRawAuthKey(c.authKeyID) c.beginTerminalShutdown() c.closeTransport() return nil @@ -1037,10 +1051,7 @@ func (s *Server) executeInboundPlan(ctx context.Context, cs *connState, c *Conn, if owner, _ := item.payload.(*rpcResultOwnerLease); owner != nil { owner.CompleteExecution(false) } - if err := s.sendResult(ctx, c, item.msgID, &mt.RPCError{ - ErrorCode: 420, - ErrorMessage: "FLOOD_WAIT_1", - }); err != nil { + if err := s.sendResult(ctx, c, item.msgID, rpcWorkerBusyError()); err != nil { return err } case inboundItemRPCAdmissionError: diff --git a/internal/mtprotoedge/layer_admission_budget_test.go b/internal/mtprotoedge/layer_admission_budget_test.go index 0a4e9d3b..97db04a0 100644 --- a/internal/mtprotoedge/layer_admission_budget_test.go +++ b/internal/mtprotoedge/layer_admission_budget_test.go @@ -2,7 +2,6 @@ package mtprotoedge import ( "context" - "errors" "reflect" "strings" "sync/atomic" @@ -93,22 +92,6 @@ type countingLayerRPCAdmission struct { decodeCalls atomic.Int32 } -type failingReplayLayerRPC struct { - LayerRPCHandler - err error -} - -func (h *failingReplayLayerRPC) PrepareAdmittedReplay( - context.Context, - [8]byte, - int64, - int64, - uint64, - tlprofile.Admission, -) (func() error, error) { - return nil, h.err -} - func (h *countingLayerRPCAdmission) AdmitLayer(profile tlprofile.Profile, b *bin.Buffer, limits tlprofile.Limits) (tlprofile.Admission, error) { h.decodeCalls.Add(1) return h.LayerRPCHandler.AdmitLayer(profile, b, limits) @@ -509,7 +492,7 @@ func TestLayerRPCAdmissionCompletedReplayReleasesWholeProvisionalBatch(t *testin if !claim.owner.CompleteExecution(true) { t.Fatal("complete replay business outcome failed") } - s.rpcResults.Put(c.authKeyID, c.sessionID, 100, &encodedOutboundMessage{body: []byte{1, 2, 3, 4}}) + storeLogicalRPCResultForTest(t, s, c, 100, &encodedOutboundMessage{body: []byte{1, 2, 3, 4}}) plan := &inboundPlan{items: []inboundItem{{kind: inboundItemRPC, msgID: 100, body: body}}} defer plan.close() @@ -530,54 +513,6 @@ func TestLayerRPCAdmissionCompletedReplayReleasesWholeProvisionalBatch(t *testin } } -func TestLayerRPCAdmissionReplayPreparationErrorIsNotSilentlyDelivered(t *testing.T) { - router := rpc.New(rpc.Config{DC: 2}, rpc.Deps{}, zaptest.NewLogger(t), clock.System) - prepareErr := errors.New("invalid replay wrapper metadata") - s := New(Options{DC: 2, LayerRPC: &failingReplayLayerRPC{ - LayerRPCHandler: router, - err: prepareErr, - }}) - c := &Conn{authKeyID: [8]byte{8, 9}, sessionID: 89, metrics: NopMetrics{}} - c.startInboundRPCScheduler(s.rpcScheduler, 1, 2, time.Second) - if err := c.FreezeLayerProfile(tlprofile.Profile225); err != nil { - t.Fatal(err) - } - body := exactOutboundLayerRPCBody(t, tlprofile.Profile225, &tg.HelpGetConfigRequest{}) - identityBuffer := &bin.Buffer{Buf: append([]byte(nil), body...)} - request, err := router.AdmitLayer(tlprofile.Profile225, identityBuffer, tlprofile.Limits{}) - if err != nil { - t.Fatal(err) - } - claim, err := s.rpcResults.AcquireIdentified(c.authKeyID, c.sessionID, 100, request.Prepared().Identity()) - if err != nil || claim.owner == nil { - t.Fatalf("completed replay owner = %v, %v", claim.owner, err) - } - if !claim.owner.CompleteExecution(true) { - t.Fatal("complete replay business outcome failed") - } - s.rpcResults.Put(c.authKeyID, c.sessionID, 100, &encodedOutboundMessage{body: []byte{1, 2, 3, 4}}) - - plan := &inboundPlan{items: []inboundItem{{kind: inboundItemRPC, msgID: 100, body: body}}} - if err := s.prepareInboundLayerRPCBatch(context.Background(), c, plan); !errors.Is(err, prepareErr) { - plan.close() - t.Fatalf("replay preparation error = %v, want %v", err, prepareErr) - } - if plan.items[0].kind == inboundItemReplayRPC { - plan.close() - t.Fatal("invalid replay metadata was converted into a deliverable cached result") - } - plan.close() - if got := c.inflightRPCBytes.Load(); got != 0 || c.rpcReserved != 0 { - t.Fatalf("failed replay preparation leaked connection budget bytes:%d tasks:%d", got, c.rpcReserved) - } - s.rpcScheduler.budgetMu.Lock() - globalTasks, globalBytes := s.rpcScheduler.tasks, s.rpcScheduler.bytes - s.rpcScheduler.budgetMu.Unlock() - if globalTasks != 0 || globalBytes != 0 { - t.Fatalf("failed replay preparation leaked global budget %d/%d", globalTasks, globalBytes) - } -} - func TestLayerRPCAdmissionTransferredBatchClosesWithoutLeak(t *testing.T) { router := rpc.New(rpc.Config{DC: 2}, rpc.Deps{}, zaptest.NewLogger(t), clock.System) s := New(Options{DC: 2, LayerRPC: router}) diff --git a/internal/mtprotoedge/layer_rpc_execution_test.go b/internal/mtprotoedge/layer_rpc_execution_test.go index 8e0ef6a4..8ff6d20a 100644 --- a/internal/mtprotoedge/layer_rpc_execution_test.go +++ b/internal/mtprotoedge/layer_rpc_execution_test.go @@ -138,6 +138,9 @@ func TestProjectionFailureCachesInternalWithoutRepeatingBusiness(t *testing.T) { if rpcErr.ErrorCode != 500 || rpcErr.ErrorMessage != "INTERNAL" { t.Fatalf("projection terminal = %+v", rpcErr) } + if _, ok := s.rpcResults.Get(c.authKeyID, c.sessionID, reqMsgID); !ok { + t.Fatal("projection receipt disappeared before duplicate acquire") + } // A same-msg replay is served from the completed exact identity; there is no // second DispatchAdmitted call even though projection failed after business // success. @@ -145,9 +148,14 @@ func TestProjectionFailureCachesInternalWithoutRepeatingBusiness(t *testing.T) { c.authKeyID, c.sessionID, reqMsgID, tlprofile.Profile227, request.Prepared().Identity(), ) - if err != nil || replay.state != rpcResultAcquireCompleted || replay.encoded != completed.encoded { + if err != nil || replay.state != rpcResultAcquireCompleted { t.Fatalf("projection replay = state:%d err:%v", replay.state, err) } + if replay.encoded == nil || replay.encoded.replayMsgID != completed.encoded.replayMsgID || + replay.encoded.replaySeqNo != completed.encoded.replaySeqNo || + !sameBacking(replay.encoded.body, completed.encoded.body) { + t.Fatal("projection replay did not reference the original logical-outbox frame") + } if got := handler.calls.Load(); got != 1 { t.Fatalf("replay repeated business calls=%d", got) } diff --git a/internal/mtprotoedge/logical_session.go b/internal/mtprotoedge/logical_session.go new file mode 100644 index 00000000..9b6fe0ea --- /dev/null +++ b/internal/mtprotoedge/logical_session.go @@ -0,0 +1,186 @@ +package mtprotoedge + +import "time" + +// logicalSession is the process-local MTProto session that survives physical +// transport replacement. The outbound state is the sole owner of every +// content-related server message until msgs_ack, explicit session destruction, +// or the bounded offline-retention window expires. +type logicalSession struct { + key sessionKey + outbound *outboundState + offlineAt time.Time + businessAuthKeyID [8]byte + businessAuthResolved bool +} + +const logicalSessionOfflineTTL = 6 * time.Minute + +func (m *SessionManager) attachLogicalSession(c *Conn, budget *outboundTrackedBudget) { + if m == nil || c == nil { + return + } + key := connSessionKey(c) + m.mu.Lock() + logical := m.logicalSessions[key] + if logical == nil { + logical = &logicalSession{ + key: key, + outbound: newOutboundState(budget), + } + m.logicalSessions[key] = logical + } + logical.outbound.persistent.Store(true) + if businessAuthKeyID, resolved := c.BusinessAuthKeyID(); resolved { + logical.businessAuthKeyID = businessAuthKeyID + logical.businessAuthResolved = true + } + logical.offlineAt = time.Time{} + c.outboundState = logical.outbound + m.mu.Unlock() +} + +// adoptLogicalSession is a construction-test/embedded-Conn bridge. Production +// Conns are attached before their actor starts; direct Conn builders may already +// own an actor-local state when a Server terminal callback publishes the receipt. +func (m *SessionManager) adoptLogicalSession(c *Conn) { + if m == nil || c == nil || c.outboundState == nil { + return + } + key := connSessionKey(c) + m.mu.Lock() + logical := m.logicalSessions[key] + if logical == nil { + logical = &logicalSession{key: key, outbound: c.outboundState} + m.logicalSessions[key] = logical + } + logical.outbound.persistent.Store(true) + if businessAuthKeyID, resolved := c.BusinessAuthKeyID(); resolved { + logical.businessAuthKeyID = businessAuthKeyID + logical.businessAuthResolved = true + } + logical.offlineAt = time.Time{} + // The actor's state pointer is immutable after startOutbound. Production + // attaches before start; this adoption bridge only marks that already-owned + // state persistent and must never write the Conn field concurrently. + m.mu.Unlock() +} + +func (m *SessionManager) markLogicalSessionOfflineLocked(key sessionKey, now time.Time) { + logical := m.logicalSessions[key] + if logical == nil || m.bySession[key] != nil || m.claims[key] != nil { + return + } + if logical.offlineAt.IsZero() { + logical.offlineAt = now + } +} + +func (m *SessionManager) destroyLogicalSessionLocked(key sessionKey) *outboundState { + logical := m.logicalSessions[key] + if logical == nil { + return nil + } + delete(m.logicalSessions, key) + return logical.outbound +} + +func (m *SessionManager) bindLogicalSessionAuthKeyLocked(key sessionKey, businessAuthKeyID [8]byte) { + if logical := m.logicalSessions[key]; logical != nil { + logical.businessAuthKeyID = businessAuthKeyID + logical.businessAuthResolved = true + } +} + +func (m *SessionManager) releaseLogicalSession(key sessionKey, state *outboundState) { + if state != nil { + state.releaseAll() + } + m.mu.RLock() + hook := m.logicalSessionReleased + m.mu.RUnlock() + if hook != nil { + hook(key) + } +} + +// ForgetLogicalSessionsForRawAuthKey is the terminal auth-key destruction path. +// Once destroy_auth_key_ok is on the wire the key can never reconnect to ACK or +// replay an old answer, so retaining any session payload or receipt is useless. +func (m *SessionManager) ForgetLogicalSessionsForRawAuthKey(authKeyID [8]byte) { + if m == nil { + return + } + var release []*logicalSession + m.mu.Lock() + for key, logical := range m.logicalSessions { + if key.authKeyID != authKeyID { + continue + } + delete(m.logicalSessions, key) + if logical != nil { + release = append(release, logical) + } + } + m.mu.Unlock() + for _, logical := range release { + m.releaseLogicalSession(logical.key, logical.outbound) + } +} + +func (m *SessionManager) sweepLogicalSessions(now time.Time) { + if m == nil { + return + } + var release []*logicalSession + m.mu.Lock() + for key, logical := range m.logicalSessions { + if logical == nil || logical.offlineAt.IsZero() || now.Sub(logical.offlineAt) < logicalSessionOfflineTTL { + continue + } + if m.bySession[key] != nil || m.claims[key] != nil { + logical.offlineAt = time.Time{} + continue + } + delete(m.logicalSessions, key) + release = append(release, logical) + } + m.mu.Unlock() + for _, logical := range release { + m.releaseLogicalSession(logical.key, logical.outbound) + } +} + +func (m *SessionManager) releaseAllLogicalSessions() { + if m == nil { + return + } + var release []*logicalSession + m.mu.Lock() + for key, logical := range m.logicalSessions { + delete(m.logicalSessions, key) + if logical != nil { + release = append(release, logical) + } + } + m.mu.Unlock() + for _, logical := range release { + m.releaseLogicalSession(logical.key, logical.outbound) + } +} + +// rpcResult returns the exact unacknowledged wire result owned by the logical +// session. The receipt ledger calls this only after validating request identity. +func (m *SessionManager) rpcResult(authKeyID [8]byte, sessionID, reqMsgID int64) (*encodedOutboundMessage, bool) { + if m == nil || reqMsgID == 0 { + return nil, false + } + key := sessionKey{authKeyID: authKeyID, sessionID: sessionID} + m.mu.RLock() + logical := m.logicalSessions[key] + m.mu.RUnlock() + if logical == nil || logical.outbound == nil { + return nil, false + } + return logical.outbound.rpcResult(reqMsgID) +} diff --git a/internal/mtprotoedge/logical_session_test.go b/internal/mtprotoedge/logical_session_test.go new file mode 100644 index 00000000..6af04d19 --- /dev/null +++ b/internal/mtprotoedge/logical_session_test.go @@ -0,0 +1,281 @@ +package mtprotoedge + +import ( + "bytes" + "encoding/binary" + "testing" + "time" + + "github.com/iamxvbaba/td/proto" + "go.uber.org/zap" +) + +// storeLogicalRPCResultForTest publishes the same production invariant as the +// outbound actor: the exact rpc_result frame enters the logical-session outbox +// before its metadata receipt becomes visible. Tests that call the receipt +// ledger directly must not invent the old impossible state where a replay row +// exists without an unacknowledged server frame. +func storeLogicalRPCResultForTest( + t *testing.T, + s *Server, + c *Conn, + reqMsgID int64, + encoded *encodedOutboundMessage, +) { + t.Helper() + if s == nil || c == nil || encoded == nil || len(encoded.body) == 0 { + t.Fatal("invalid logical rpc_result fixture") + } + if c.outboundState == nil { + s.conns.attachLogicalSession(c, s.outboundTrackedBudget) + } else { + s.conns.adoptLogicalSession(c) + } + physicalReqMsgID := encoded.writtenRequestID() + if physicalReqMsgID == 0 { + physicalReqMsgID = reqMsgID + } + frameBody := encoded.body + if physicalReqMsgID != reqMsgID && encoded.typeID == proto.ResultTypeID { + physical, err := cloneRPCResultForRequest(encoded, physicalReqMsgID, true) + if err != nil { + t.Fatalf("retarget logical rpc_result fixture: %v", err) + } + frameBody = physical.body + } + state := c.outboundState + state.mu.Lock() + if state.budget == nil || !state.budget.reserve(len(frameBody)) { + state.mu.Unlock() + t.Fatal("reserve logical rpc_result fixture") + } + candidate := reqMsgID*4 + 1 + if candidate <= 0 { + candidate = 1 + } + frame := &outboundFrame{ + msgID: state.reserveMsgID(candidate), seqNo: state.peekSeqNo(true), + typeID: proto.ResultTypeID, body: frameBody, reqMsgID: physicalReqMsgID, + priority: encoded.priority, delivery: encoded.delivery, + compressed: encoded.compressed, uncompressedBytes: encoded.uncompressedBytes, + layer: encoded.layer, layerInvariant: encoded.layerInvariant, + reservedBytes: len(frameBody), reservationBudget: state.budget, + } + if err := state.admitReserved(frame); err != nil { + frame.releaseReservation(state.budget) + state.mu.Unlock() + t.Fatalf("admit logical rpc_result fixture: %v", err) + } + encoded.replayMsgID = frame.msgID + encoded.replaySeqNo = frame.seqNo + state.mu.Unlock() + s.rpcResults.Put(c.authKeyID, c.sessionID, reqMsgID, encoded) +} + +func acknowledgeLogicalRPCResultForTest(t *testing.T, s *Server, c *Conn, reqMsgID int64) { + t.Helper() + if s == nil || c == nil || c.outboundState == nil { + t.Fatal("missing logical rpc_result fixture") + } + state := c.outboundState + state.mu.Lock() + msgID := state.byRequest[reqMsgID] + requestIDs := state.ack([]int64{msgID}) + state.mu.Unlock() + if len(requestIDs) != 1 || requestIDs[0] != reqMsgID { + t.Fatalf("logical ACK resolved request ids %v", requestIDs) + } + if !s.rpcResults.Acknowledge(c.authKeyID, c.sessionID, reqMsgID) { + t.Fatal("logical ACK did not release result receipt") + } +} + +func TestLogicalSessionOwnsExactResultAcrossPhysicalReconnect(t *testing.T) { + manager := NewSessionManager(zap.NewNop()) + budget := newOutboundTrackedBudget(1024) + authKeyID := [8]byte{1, 2, 3, 4, 5, 6, 7, 8} + first := &Conn{authKeyID: authKeyID, sessionID: 42} + manager.attachLogicalSession(first, budget) + + body := append([]byte{0xf3, 0x5c, 0x6d, 0xf3}, make([]byte, 28)...) + binary.LittleEndian.PutUint64(body[4:12], uint64(77)) + if !budget.reserve(len(body)) { + t.Fatal("reserve result body") + } + frame := &outboundFrame{ + msgID: 101, seqNo: 1, typeID: proto.ResultTypeID, body: body, + reqMsgID: 77, reservedBytes: len(body), reservationBudget: budget, + } + first.outboundState.mu.Lock() + if err := first.outboundState.admitReserved(frame); err != nil { + first.outboundState.mu.Unlock() + t.Fatalf("admit result: %v", err) + } + first.outboundState.mu.Unlock() + + second := &Conn{authKeyID: authKeyID, sessionID: 42} + manager.attachLogicalSession(second, budget) + if first.outboundState != second.outboundState { + t.Fatal("physical reconnect did not reuse logical outbound state") + } + replay, ok := manager.rpcResult(authKeyID, 42, 77) + if !ok { + t.Fatal("logical result not found after reconnect") + } + if replay.replayMsgID != frame.msgID || replay.replaySeqNo != frame.seqNo || !bytes.Equal(replay.body, body) { + t.Fatalf("replay identity/body = msg:%d seq:%d bytes:%d", replay.replayMsgID, replay.replaySeqNo, len(replay.body)) + } + if &replay.body[0] != &frame.body[0] { + t.Fatal("replay created a second payload owner") + } + attempt, err := cloneRPCResultForRequest(replay, replay.reqMsgID, false) + if err != nil { + t.Fatalf("clone same-request replay descriptor: %v", err) + } + if attempt.replayMsgID != frame.msgID || attempt.replaySeqNo != frame.seqNo || + !sameBacking(attempt.body, frame.body) { + t.Fatal("queued duplicate lost stable logical frame identity") + } +} + +func TestLogicalSessionACKReleasesPayloadAndReceipt(t *testing.T) { + manager := NewSessionManager(zap.NewNop()) + budget := newOutboundTrackedBudget(1024) + authKeyID := [8]byte{8, 7, 6, 5, 4, 3, 2, 1} + c := &Conn{authKeyID: authKeyID, sessionID: 43} + manager.attachLogicalSession(c, budget) + + body := make([]byte, 64) + if !budget.reserve(len(body)) { + t.Fatal("reserve result body") + } + frame := &outboundFrame{ + msgID: 201, seqNo: 1, typeID: proto.ResultTypeID, body: body, + reqMsgID: 88, reservedBytes: len(body), reservationBudget: budget, + } + c.outboundState.mu.Lock() + if err := c.outboundState.admitReserved(frame); err != nil { + c.outboundState.mu.Unlock() + t.Fatalf("admit result: %v", err) + } + c.outboundState.mu.Unlock() + + ledger := newRPCResultCacheWithFairCapacity(time.Now, rpcResultCacheCapacity{ + maxPending: 16, maxPendingPerAuth: 16, + globalMaxEntries: 16, globalMaxBytes: 16, + authMaxEntries: 16, authMaxBytes: 16, + sessionMaxEntries: 16, sessionMaxBytes: 16, + sessions: manager, + }) + claim, err := ledger.Acquire(authKeyID, 43, 88) + if err != nil || claim.state != rpcResultAcquireOwner { + t.Fatalf("acquire owner = %#v, %v", claim, err) + } + claim.owner.CompleteExecution(true) + claim.owner.HandOff() + ledger.Put(authKeyID, 43, 88, &encodedOutboundMessage{body: body, typeID: proto.ResultTypeID, reqMsgID: 88}) + + key := rpcResultCacheKey{authKeyID: authKeyID, sessionID: 43, reqMsgID: 88} + shard := ledger.shard(key) + shard.mu.Lock() + entry := shard.byKey[key].Value.(*rpcResultCacheEntry) + if entry.encoded != nil || entry.size != 1 { + shard.mu.Unlock() + t.Fatalf("completed ledger retained payload: encoded=%p size=%d", entry.encoded, entry.size) + } + shard.mu.Unlock() + + c.outboundState.mu.Lock() + acked := c.outboundState.ack([]int64{201}) + c.outboundState.mu.Unlock() + if len(acked) != 1 || acked[0] != 88 { + t.Fatalf("acked request ids = %v", acked) + } + if !ledger.Acknowledge(authKeyID, 43, 88) { + t.Fatal("ledger did not observe ACK") + } + if got := budget.snapshot(); got != 0 { + t.Fatalf("tracked payload bytes after ACK = %d", got) + } + shard.mu.Lock() + _, retained := shard.byKey[key] + shard.mu.Unlock() + if retained { + t.Fatal("ACK retained a completed receipt") + } +} + +func TestLogicalSessionCapacityNeverEvictsUnackedFrame(t *testing.T) { + budget := newOutboundTrackedBudget(1024) + state := newOutboundStateWithLimits(budget, 1, 16) + firstBody := make([]byte, 8) + if !budget.reserve(len(firstBody)) { + t.Fatal("reserve first body") + } + first := &outboundFrame{ + msgID: 1, seqNo: 1, typeID: proto.ResultTypeID, body: firstBody, + reqMsgID: 11, reservedBytes: len(firstBody), reservationBudget: budget, + } + if err := state.admitReserved(first); err != nil { + t.Fatalf("admit first: %v", err) + } + second := &outboundFrame{msgID: 2, seqNo: 3, typeID: proto.ResultTypeID, body: make([]byte, 8), reqMsgID: 12} + if err := state.admitReserved(second); err == nil { + t.Fatal("capacity admitted a second unacknowledged frame") + } + if state.pending[first.msgID] != first || state.byRequest[first.reqMsgID] != first.msgID { + t.Fatal("capacity failure evicted or rewired the existing frame") + } + state.releaseAll() +} + +func TestLogicalSessionDestroyReleasesPayloadAndCompletedReceipt(t *testing.T) { + s := New(Options{}) + authKeyID := [8]byte{9, 1} + c := &Conn{authKeyID: authKeyID, sessionID: 91} + claim, err := s.rpcResults.Acquire(authKeyID, 91, 901) + if err != nil || claim.owner == nil { + t.Fatalf("acquire owner: %#v err=%v", claim, err) + } + claim.owner.CompleteExecution(true) + storeLogicalRPCResultForTest(t, s, c, 901, &encodedOutboundMessage{ + body: make([]byte, 32), typeID: proto.ResultTypeID, reqMsgID: 901, + }) + if _, ok := s.rpcResults.Get(authKeyID, 91, 901); !ok { + t.Fatal("logical result fixture was not published") + } + if removed := s.conns.DestroySessionForAuthKey(authKeyID, 91); removed { + t.Fatal("construction-only logical session unexpectedly reported active") + } + if _, ok := s.rpcResults.Get(authKeyID, 91, 901); ok { + t.Fatal("destroy retained completed receipt") + } + if got := s.outboundTrackedBudget.snapshot(); got != 0 { + t.Fatalf("destroy retained %d payload bytes", got) + } +} + +func TestBusinessAuthRevocationReleasesOfflineTempLogicalSession(t *testing.T) { + s := New(Options{}) + rawAuthKeyID := [8]byte{9, 2} + businessAuthKeyID := [8]byte{9, 3} + c := &Conn{authKeyID: rawAuthKeyID, sessionID: 92} + c.SetBusinessAuthKeyID(businessAuthKeyID) + claim, err := s.rpcResults.Acquire(rawAuthKeyID, 92, 902) + if err != nil || claim.owner == nil { + t.Fatalf("acquire owner: %#v err=%v", claim, err) + } + claim.owner.CompleteExecution(true) + storeLogicalRPCResultForTest(t, s, c, 902, &encodedOutboundMessage{ + body: make([]byte, 48), typeID: proto.ResultTypeID, reqMsgID: 902, + }) + if closed := s.conns.CloseSessionsForBusinessAuthKey(businessAuthKeyID); closed != 0 { + t.Fatalf("offline logical revocation closed %d physical conns", closed) + } + if _, ok := s.rpcResults.Get(rawAuthKeyID, 92, 902); ok { + t.Fatal("business auth revocation retained temp-key receipt") + } + if got := s.outboundTrackedBudget.snapshot(); got != 0 { + t.Fatalf("business auth revocation retained %d payload bytes", got) + } +} diff --git a/internal/mtprotoedge/outbound.go b/internal/mtprotoedge/outbound.go index 18b26343..e21eb4f3 100644 --- a/internal/mtprotoedge/outbound.go +++ b/internal/mtprotoedge/outbound.go @@ -50,8 +50,8 @@ const ( maxTrackedServerMsgIDs = 4096 maxTrackedAckedMsgIDs = 1024 // maxTrackedServerBytes 是 pending（已发送待 ack、用于 resend）总 body 字节上限。 - // 与 maxTrackedServerMsgIDs 并列：客户端从不 ack 时，大响应体按字节滚动丢弃， - // 防 pending 被「4096 条 × 大 body」撑爆。 + // 与 maxTrackedServerMsgIDs 并列；到达任一上限后拒绝新可靠 frame，绝不滚动 + // 丢弃尚未 ACK 的旧 frame。 maxTrackedServerBytes = 64 << 20 // 64 MiB // Encrypted transport adds auth-key/msg-key, plaintext headers, randomized // padding and codec framing. Reject before creating the two encryption buffers. @@ -214,6 +214,12 @@ type encodedOutboundMessage struct { layer *outboundLayerBinding compressed bool uncompressedBytes int + // replayMsgID/replaySeqNo identify an existing logical-session frame. They + // are populated only by the receipt ledger's outbox lookup, never by a newly + // encoded result. A replay writes that exact frame instead of allocating a + // second payload owner or a new MTProto message identity. + replayMsgID int64 + replaySeqNo int32 } type rpcResultDeliveryState uint32 @@ -263,7 +269,7 @@ type rpcResultDeliveryCoordinator struct { // deferredToReplay is sticky while a successful initConnection retarget owns // the logical hook. The ordinary source terminal may mark physical delivery, // but only the alias restore barrier may claim the hook. On terminal alias - // failure the flag is released so a later completed-cache replay can retry. + // failure the flag is released so a later logical-outbox replay can retry. deferredToReplay bool } @@ -722,18 +728,25 @@ func cloneRPCResultForRequest(encoded *encodedOutboundMessage, reqMsgID int64, s if shareDelivery { delivery = encoded.delivery } + var replayMsgID int64 + var replaySeqNo int32 + if reqMsgID == encoded.reqMsgID { + replayMsgID = encoded.replayMsgID + replaySeqNo = encoded.replaySeqNo + } return &encodedOutboundMessage{ body: body, typeID: encoded.typeID, reqMsgID: reqMsgID, priority: encoded.priority, delivery: delivery, compressed: encoded.compressed, layer: encoded.layer, layerInvariant: encoded.layerInvariant, uncompressedBytes: encoded.uncompressedBytes, + replayMsgID: replayMsgID, replaySeqNo: replaySeqNo, }, nil } // cloneRPCResultForRequestReserved charges the target connection's retained-body // budget before a retarget copy can exist. Even the same-req_id zero-copy case -// needs a reservation: the replay/rewrap owner may outlive cache eviction while -// it is queued, so its immutable body must remain independently pinned. +// needs a reservation: the replay/rewrap attempt may outlive ACK/removal of its +// source frame while queued, so its immutable body must remain independently pinned. func (c *Conn) cloneRPCResultForRequestReserved( encoded *encodedOutboundMessage, reqMsgID int64, @@ -776,14 +789,20 @@ type outboundFrame struct { // but their bytes must remain on the independent control budget for the full lifetime. reservationBudget *outboundTrackedBudget reqMsgID int64 + priority outboundPriority + delivery *rpcResultDelivery + compressed bool + uncompressedBytes int // layer is retained only for proactive session-bound frames so a later // msg_resend_req cannot replay bytes from an obsolete profile epoch. - layer *outboundLayerBinding - sentAt time.Time - sends int + layer *outboundLayerBinding + layerInvariant bool + sentAt time.Time + sends int } type outboundState struct { + mu sync.Mutex pending map[int64]*outboundFrame order []int64 byRequest map[int64]int64 @@ -793,6 +812,16 @@ type outboundState struct { maxMessages int maxBytes int budget *outboundTrackedBudget + // sentContentMessages is logical-session state, not physical-connection + // state. Reserving a new content frame advances it before the first write so + // a failed write can be replayed with the same seq_no after reconnect. + sentContentMessages int32 + lastMsgID int64 + // persistent becomes true once this state is owned by a logical session. + // Directly constructed Conns may start with actor-local ownership and be + // adopted only from the terminal callback after a failed first write; the + // actor must not release that outbox while the logical session retains it. + persistent atomic.Bool } // outboundTrackedBudget 是 body/control/write 三类预算共用的原子 byte-budget primitive。 @@ -1001,6 +1030,12 @@ func (c *Conn) startOutbound() { c.outboundBulk = make(chan outboundOp, bulkSize) c.outboundStop = make(chan struct{}) c.outboundDone = make(chan struct{}) + // Publish the actor state before starting its goroutine. The pointer remains + // immutable for the physical Conn lifetime; logical-session adoption may only + // mark the existing state persistent. + if c.outboundState == nil { + c.outboundState = newOutboundState(c.outboundTrackedBudget) + } go c.outboundLoop() } @@ -1669,12 +1704,14 @@ func (c *Conn) endOutboundEnqueue() { } func (c *Conn) outboundLoop() { - state := newOutboundState(c.outboundTrackedBudget) + state := c.outboundState ordinarySinceBulk := 0 defer func() { - // pending frames belong exclusively to this actor. Releasing after drain ensures no - // resend path can race the final budget return and no Conn body survives actor exit. - state.releaseAll() + // A Server Conn leaves pending frames with the logical session. Standalone + // construction/tests retain the old actor-local ownership boundary. + if !state.persistent.Load() { + state.releaseAll() + } close(c.outboundDone) }() for { @@ -1790,29 +1827,39 @@ func (c *Conn) handleOutboundOp(state *outboundState, op outboundOp) { if op.kind != outboundSend { defer op.releaseReservation(state.budget) } + // Physical generations may overlap briefly during activation fencing. The + // logical-session mutex preserves one writer/seq/outbox state machine across + // both actors without transferring payload ownership. Terminal callbacks run + // only after unlock: publication may look the frame up in this same outbox. + state.mu.Lock() + var ( + result outboundResult + acked []int64 + ) switch op.kind { case outboundSend: - c.handleOutboundSend(state, op) + result.err = c.handleOutboundSend(state, op) case outboundAck: - for _, reqMsgID := range state.ack(op.ids) { - if c.rpcResultAcked != nil { - c.rpcResultAcked(c, reqMsgID) - } - } + acked = state.ack(op.ids) case outboundQueryState: - op.finish(outboundResult{info: state.stateInfo(op.ids)}) + result.info = state.stateInfo(op.ids) case outboundResend: - info, err := c.handleOutboundResend(state, op.ctx, op.ids) - op.finish(outboundResult{info: info, err: err}) + result.info, result.err = c.handleOutboundResend(state, op.ctx, op.ids) case outboundResendByRequest: - resent, err := c.handleOutboundResendByRequest(state, op.ctx, op.reqMsgID) - op.finish(outboundResult{resent: resent, err: err}) + result.resent, result.err = c.handleOutboundResendByRequest(state, op.ctx, op.reqMsgID) default: - op.finish(outboundResult{err: fmt.Errorf("unknown outbound op %d", op.kind)}) + result.err = fmt.Errorf("unknown outbound op %d", op.kind) } + state.mu.Unlock() + for _, reqMsgID := range acked { + if c.rpcResultAcked != nil { + c.rpcResultAcked(c, reqMsgID) + } + } + op.finish(result) } -func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) { +func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) error { var binding *outboundLayerBinding if op.encoded != nil { binding = op.encoded.layer @@ -1861,7 +1908,7 @@ func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) { } var frame *outboundFrame if err == nil { - frame, err = c.buildFrame(op.ctx, op.msgType, op.msg, op.encoded) + frame, err = c.buildFrameWithState(op.ctx, op.msgType, op.msg, op.encoded, state) } // A profile-bound preparation can allocate a different body. Reserve the // replacement before dropping the original prepared-body reservation. The @@ -1875,11 +1922,30 @@ func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) { reserved = len(frame.body) } } - if err == nil && frame != nil && frameNeedsAck(frame.typeID) { - // The queue reservation is transferred to pending after write. A frame larger - // than the per-Conn resend ceiling is rejected before any bytes hit the wire. + needsAck := err == nil && frame != nil && frameNeedsAck(frame.typeID) + replaying := false + if needsAck && frame.replayMsgID() != 0 { + if existing := state.pending[frame.msgID]; existing != nil { + frame = existing + replaying = true + } + } + if needsAck && !replaying { + // Transfer the producer reservation into the logical outbox before the + // first physical write. A write failure therefore remains replayable on a + // replacement connection with the same msg_id/seq_no. if len(frame.body) > maxTrackedServerBytes { err = ErrOutboundTrackedBudget + } else { + frame.reservedBytes = reserved + frame.reservationBudget = reservationBudget + if admitErr := state.admitReserved(frame); admitErr != nil { + frame.reservedBytes = 0 + frame.reservationBudget = nil + err = admitErr + } else { + reserved = 0 + } } } if errors.Is(err, ErrOutboundTrackedBudget) { @@ -1894,18 +1960,6 @@ func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) { if err == nil { err = c.writeFrame(op.ctx, frame) } - if err == nil && frame != nil && frameNeedsAck(frame.typeID) { - // 写成功后才提交 content seq_no 递增（peekSeqNo 已按当前计数算好本帧 seq_no）。 - c.commitContentSeqNo() - frame.reservedBytes = reserved - frame.reservationBudget = reservationBudget - reserved = 0 - if dropped := state.addReserved(frame); dropped > 0 { - for i := 0; i < dropped; i++ { - c.metrics.OutboundDropped("tracked_queue_overflow") - } - } - } queueWait := time.Since(op.enqueuedAt) bytes := 0 typeID := uint32(0) @@ -1914,7 +1968,7 @@ func (c *Conn) handleOutboundSend(state *outboundState, op outboundOp) { typeID = frame.typeID } c.metrics.OutboundSend(typeID, queueWait, bytes, err) - op.finish(outboundResult{err: err}) + return err } func (c *Conn) handleOutboundResend(state *outboundState, ctx context.Context, ids []int64) ([]byte, error) { @@ -2141,6 +2195,16 @@ func (c *Conn) ensureOutboundControlTrackedBudget() *outboundTrackedBudget { } func (c *Conn) buildFrame(ctx context.Context, t proto.MessageType, msg bin.Encoder, encoded *encodedOutboundMessage) (*outboundFrame, error) { + return c.buildFrameWithState(ctx, t, msg, encoded, c.outboundState) +} + +func (c *Conn) buildFrameWithState( + ctx context.Context, + t proto.MessageType, + msg bin.Encoder, + encoded *encodedOutboundMessage, + state *outboundState, +) (*outboundFrame, error) { if encoded == nil { var err error encoded, err = encodeOutboundMessage(msg) @@ -2155,14 +2219,29 @@ func (c *Conn) buildFrame(ctx context.Context, t proto.MessageType, msg bin.Enco return nil, ErrOutboundLayerBindingRequired } content := frameNeedsAck(encoded.typeID) - msgID := c.msgID.New(t) + msgID := encoded.replayMsgID + seqNo := encoded.replaySeqNo + if msgID == 0 { + msgID = c.msgID.New(t) + if state != nil { + msgID = state.reserveMsgID(msgID) + seqNo = state.peekSeqNo(content) + } else { + seqNo = c.peekSeqNo(content) + } + } return &outboundFrame{ - msgID: msgID, - seqNo: c.peekSeqNo(content), - typeID: encoded.typeID, - body: encoded.body, - reqMsgID: encoded.reqMsgID, - layer: encoded.layer, + msgID: msgID, + seqNo: seqNo, + typeID: encoded.typeID, + body: encoded.body, + reqMsgID: encoded.reqMsgID, + layer: encoded.layer, + layerInvariant: encoded.layerInvariant, + priority: encoded.priority, + delivery: encoded.delivery, + compressed: encoded.compressed, + uncompressedBytes: encoded.uncompressedBytes, }, nil } @@ -2545,7 +2624,18 @@ func outboundRequestMsgID(msg bin.Encoder) int64 { // addReserved 接管调用方已经取得的全局 body 预算。pending 的每个元素恰好对应一份 // reservation；后续只有 removePending/releaseAll 能归还。 -func (s *outboundState) addReserved(frame *outboundFrame) int { +func (s *outboundState) admitReserved(frame *outboundFrame) error { + if _, exists := s.pending[frame.msgID]; exists { + return fmt.Errorf("mtprotoedge: duplicate outbound msg_id inserted into resend tracking") + } + if len(s.pending) >= s.maxMessages || s.totalBytes > s.maxBytes-len(frame.body) { + return ErrOutboundTrackedBudget + } + s.insertReserved(frame) + return nil +} + +func (s *outboundState) insertReserved(frame *outboundFrame) { if _, exists := s.pending[frame.msgID]; exists { panic("mtprotoedge: duplicate outbound msg_id inserted into resend tracking") } @@ -2561,6 +2651,15 @@ func (s *outboundState) addReserved(frame *outboundFrame) int { } s.byRequest[frame.reqMsgID] = frame.msgID } + if frameNeedsAck(frame.typeID) { + s.sentContentMessages++ + } +} + +// addReserved is retained as a focused-test helper. Production uses +// admitReserved and never evicts an unacknowledged frame. +func (s *outboundState) addReserved(frame *outboundFrame) int { + s.insertReserved(frame) return s.shrinkPending() } @@ -2619,6 +2718,8 @@ func (s *outboundState) markAcked(id int64) { } } +// shrinkPending remains only for focused legacy state tests. Production +// admission never calls it: unacknowledged frames must not be silently evicted. func (s *outboundState) shrinkPending() int { dropped := 0 for (len(s.pending) > s.maxMessages || s.totalBytes > s.maxBytes) && len(s.order) > 0 { @@ -2652,6 +2753,8 @@ func (s *outboundState) removePending(id int64) bool { } func (s *outboundState) releaseAll() { + s.mu.Lock() + defer s.mu.Unlock() for _, frame := range s.pending { frame.body = nil frame.releaseReservation(s.budget) @@ -2662,6 +2765,58 @@ func (s *outboundState) releaseAll() { s.totalBytes = 0 } +func (s *outboundState) peekSeqNo(content bool) int32 { + seqNo := s.sentContentMessages * 2 + if content { + seqNo++ + } + return seqNo +} + +func (s *outboundState) reserveMsgID(candidate int64) int64 { + if candidate <= s.lastMsgID { + candidate = s.lastMsgID + 4 + } + for s.pending[candidate] != nil { + candidate += 4 + } + s.lastMsgID = candidate + return candidate +} + +func (s *outboundState) rpcResult(reqMsgID int64) (*encodedOutboundMessage, bool) { + if s == nil || reqMsgID == 0 { + return nil, false + } + s.mu.Lock() + defer s.mu.Unlock() + msgID := s.byRequest[reqMsgID] + frame := s.pending[msgID] + if frame == nil || frame.typeID != proto.ResultTypeID || len(frame.body) == 0 { + return nil, false + } + return &encodedOutboundMessage{ + body: frame.body, + typeID: frame.typeID, + reqMsgID: frame.reqMsgID, + priority: frame.priority, + delivery: frame.delivery, + layer: frame.layer, + layerInvariant: frame.layerInvariant, + compressed: frame.compressed, + uncompressedBytes: frame.uncompressedBytes, + replayMsgID: frame.msgID, + replaySeqNo: frame.seqNo, + }, true +} + +func (f *outboundFrame) replayMsgID() int64 { + if f == nil { + return 0 + } + return f.msgID +} + func (f *outboundFrame) releaseReservation(defaultBudget *outboundTrackedBudget) { if f == nil || f.reservedBytes <= 0 { return diff --git a/internal/mtprotoedge/rpc_capacity_error.go b/internal/mtprotoedge/rpc_capacity_error.go new file mode 100644 index 00000000..8ce3b74c --- /dev/null +++ b/internal/mtprotoedge/rpc_capacity_error.go @@ -0,0 +1,20 @@ +package mtprotoedge + +import "github.com/iamxvbaba/td/mt" + +const ( + rpcWorkerBusyErrorCode = 500 + rpcWorkerBusyErrorMessage = "WORKER_BUSY_TOO_LONG_RETRY" +) + +// rpcWorkerBusyError reports transient server admission pressure. A local +// worker/queue/ledger ceiling is not Telegram flood control: returning a 420 +// FLOOD_WAIT would falsely blame the account and makes clients surface or cache +// a rate-limit state. Official clients classify 5xx as transient; TDLib and +// DrKlo additionally recognize WORKER_BUSY_TOO_LONG_RETRY and retry with delay. +func rpcWorkerBusyError() *mt.RPCError { + return &mt.RPCError{ + ErrorCode: rpcWorkerBusyErrorCode, + ErrorMessage: rpcWorkerBusyErrorMessage, + } +} diff --git a/internal/mtprotoedge/rpc_flight.go b/internal/mtprotoedge/rpc_flight.go index 35e4ff91..de57cd70 100644 --- a/internal/mtprotoedge/rpc_flight.go +++ b/internal/mtprotoedge/rpc_flight.go @@ -18,6 +18,7 @@ var ( ErrRPCResultSubscriberCapacity = errors.New("mtproto rpc result subscriber capacity exhausted") ErrRPCResultFlightInvalid = errors.New("mtproto rpc result in-flight claim is invalid") ErrRPCResultIdentityMismatch = errors.New("mtproto rpc result request identity mismatch") + ErrRPCResultReplayUnavailable = errors.New("mtproto rpc result replay payload is unavailable") ErrRPCAdmissionSeqExhausted = errors.New("mtproto rpc admission sequence exhausted") ) @@ -61,6 +62,7 @@ type rpcResultAcquireState uint8 const ( rpcResultAcquireCompleted rpcResultAcquireState = iota + 1 + rpcResultAcquireAcknowledged rpcResultAcquirePending rpcResultAcquireOwner ) @@ -70,6 +72,8 @@ const ( // // Exactly one state-specific field is non-nil: // - completed: encoded contains the immutable completed rpc_result; +// - acknowledged: the client ACKed the completed rpc_result, so only the +// compact request receipt remains and the duplicate must be ACK-only; // - pending: waiter joins the already-running owner; // - owner: owner must eventually complete through rpcResultCache.Put or Abort. type rpcResultAcquire struct { @@ -82,7 +86,7 @@ type rpcResultAcquire struct { executionOK bool } -// rpcResultFlight is not part of the completed cache TTL lifecycle. Its +// rpcResultFlight is not part of the completed receipt TTL lifecycle. Its // done channel is closed exactly once while holding the owning cache shard lock; // channel close publishes encoded/ok to all waiters without a waiter goroutine. type rpcResultFlight struct { @@ -93,6 +97,11 @@ type rpcResultFlight struct { executionDone bool executionOK bool executionSubscribers []func(bool) + // acknowledged is set only from the sole outbound actor's server-msg-id to + // request-msg-id mapping. It can win the race with the asynchronous Put; + // publication then keeps only the compact receipt while still resolving + // waiters with the immutable result that was already written on the wire. + acknowledged bool // subscriberSlots counts callbacks retained by this pending flight. Result // and execution callbacks are charged independently; a replay alias installs // both atomically so a capacity failure cannot leave half an alias behind. @@ -465,7 +474,7 @@ func (l *rpcResultFlightLimit) snapshot() int64 { // Acquire atomically returns a completed result, joins the existing in-flight // owner, or installs the unique owner lease. Pending entries have a separate -// lifecycle from completed cache TTL but reserve the same global/auth/session +// lifecycle from completed receipt TTL but reserve the same global/auth/session // ownership that Put later transfers to a completed result or tombstone. func (c *rpcResultCache) Acquire(authKeyID [8]byte, sessionID, reqMsgID int64) (rpcResultAcquire, error) { return c.acquire(authKeyID, sessionID, reqMsgID, rpcResultRequestIdentity{}) @@ -544,10 +553,35 @@ func (c *rpcResultCache) acquire( s.mu.Unlock() return rpcResultAcquire{}, identityMismatch(entry.identity) } - if entry.capacity || entry.encoded == nil { + if entry.acknowledged { + result := rpcResultAcquire{ + state: rpcResultAcquireAcknowledged, admissionSeq: entry.admissionSeq, + executionKnown: entry.executionKnown, executionOK: entry.executionOK, + } + s.mu.Unlock() + return result, nil + } + if entry.capacity { s.mu.Unlock() return rpcResultAcquire{}, ErrRPCResultFlightCapacity } + if c.sessions != nil { + result := rpcResultAcquire{ + state: rpcResultAcquireCompleted, admissionSeq: entry.admissionSeq, + executionKnown: entry.executionKnown, executionOK: entry.executionOK, + } + s.mu.Unlock() + encoded, replayable := c.sessions.rpcResult(authKeyID, sessionID, reqMsgID) + if !replayable { + return rpcResultAcquire{}, ErrRPCResultReplayUnavailable + } + result.encoded = encoded + return result, nil + } + if entry.encoded == nil { + s.mu.Unlock() + return rpcResultAcquire{}, ErrRPCResultReplayUnavailable + } result := rpcResultAcquire{ state: rpcResultAcquireCompleted, admissionSeq: entry.admissionSeq, encoded: entry.encoded, executionKnown: entry.executionKnown, executionOK: entry.executionOK, @@ -560,6 +594,14 @@ func (c *rpcResultCache) acquire( s.mu.Unlock() return rpcResultAcquire{}, identityMismatch(flight.identity) } + if flight.acknowledged { + result := rpcResultAcquire{ + state: rpcResultAcquireAcknowledged, admissionSeq: flight.admissionSeq, + executionKnown: flight.executionDone, executionOK: flight.executionOK, + } + s.mu.Unlock() + return result, nil + } result := rpcResultAcquire{ state: rpcResultAcquirePending, admissionSeq: flight.admissionSeq, @@ -621,7 +663,7 @@ func (c *rpcResultCache) acquire( } // completeRPCResultFlightLocked publishes encoded to the current owner claim. -// The caller must hold s.mu and must publish the completed cache entry first. +// The caller must hold s.mu and must publish the completed receipt first. func (c *rpcResultCache) completeRPCResultFlightLocked( s *rpcResultCacheShard, key rpcResultCacheKey, diff --git a/internal/mtprotoedge/rpc_result_cache.go b/internal/mtprotoedge/rpc_result_cache.go index 288f6bcf..9b3198c1 100644 --- a/internal/mtprotoedge/rpc_result_cache.go +++ b/internal/mtprotoedge/rpc_result_cache.go @@ -2,6 +2,7 @@ package mtprotoedge import ( "container/list" + "context" "encoding/binary" "hash/maphash" "sync" @@ -26,12 +27,6 @@ const ( rpcResultCacheSessionMaxEntries = 1 << 14 rpcResultCacheSessionMaxBytes = 16 << 20 rpcResultFlightMaxPendingPerAuth = 1 << 11 - // Keep every transport-legal rpc_result cacheable. Converting the constant - // difference to uint64 intentionally fails compilation if a future transport - // limit grows beyond the completed-result budget. - _ = uint64(rpcResultCacheMaxBytes - maxOutboundBodyBytes) - _ = uint64(rpcResultCacheAuthMaxBytes - maxOutboundBodyBytes) - _ = uint64(rpcResultCacheSessionMaxBytes - maxOutboundBodyBytes) // rpcResultCacheShards hashes the complete replay identity with a random // per-instance maphash seed. Including req_msg_id spreads one hot session's // independent requests instead of forcing them through one mutex. The shard @@ -46,7 +41,10 @@ type rpcResultCacheKey struct { } type rpcResultCacheEntry struct { - key rpcResultCacheKey + key rpcResultCacheKey + // encoded is used only by legacy focused constructors whose cache has no + // SessionManager. Production rows are metadata receipts and always leave it + // nil; exact bytes have one owner in the logical-session outbox. encoded *encodedOutboundMessage size int expiresAt time.Time @@ -54,6 +52,10 @@ type rpcResultCacheEntry struct { admissionSeq uint64 executionKnown bool executionOK bool + // acknowledged exists only for the ACK-before-Put race. A completed row is + // removed immediately on ACK; unlike the previous design, success receipts + // are not retained for the rest of the 331-second horizon. + acknowledged bool // capacity marks a bounded replay tombstone. The original owner and its // already-joined waiters received encoded, but the byte budget could not // retain that body. Keeping the immutable identity until TTL prevents a @@ -84,6 +86,10 @@ type rpcResultCache struct { flightLimit rpcResultFlightLimit subscriberBudget *rpcResultSubscriberBudget subscriberPerFlight int + // sessions is the production payload source. Completed ledger rows retain + // only request identity/outcome/TTL metadata; exact bytes stay solely in the + // logical-session unacked outbox. Nil is kept as a legacy unit-test seam. + sessions *SessionManager // nextAdmissionSeq is the process-wide ordering authority for auth-key // shared Layer defaults. Exact owners allocate once; joins/replays retain the // owner's value from their flight/completed descriptor. @@ -172,6 +178,7 @@ type rpcResultCacheCapacity struct { subscriberMaxAuth int subscriberMaxSession int subscriberMaxPerFlight int + sessions *SessionManager } func newRPCResultCacheWithFairCapacity(now func() time.Time, capacity rpcResultCacheCapacity) *rpcResultCache { @@ -214,7 +221,7 @@ func newRPCResultCacheWithFairCapacity(now func() time.Time, capacity rpcResultC if capacity.subscriberMaxPerFlight <= 0 { capacity.subscriberMaxPerFlight = rpcResultSubscriberMaxPerFlight } - c := &rpcResultCache{hashSeed: maphash.MakeSeed()} + c := &rpcResultCache{hashSeed: maphash.MakeSeed(), sessions: capacity.sessions} c.completedBytes.max = capacity.globalMaxBytes c.completedEntries.max = int64(capacity.globalMaxEntries) c.flightLimit.max = int64(capacity.maxPending) @@ -267,21 +274,62 @@ func (c *rpcResultCache) Get(authKeyID [8]byte, sessionID, reqMsgID int64) (*enc now := s.now() s.mu.Lock() - defer s.mu.Unlock() elem, ok := s.byKey[key] if !ok { + s.mu.Unlock() return nil, false } entry := elem.Value.(*rpcResultCacheEntry) if !entry.expiresAt.After(now) { s.removeElement(elem) + s.mu.Unlock() return nil, false } - if entry.capacity || entry.encoded == nil { + if entry.capacity || entry.acknowledged { + s.mu.Unlock() return nil, false } - return entry.encoded, true + if c.sessions != nil { + s.mu.Unlock() + return c.sessions.rpcResult(authKeyID, sessionID, reqMsgID) + } + encoded := entry.encoded + s.mu.Unlock() + return encoded, encoded != nil +} + +// Acknowledge removes a completed rpc_result receipt immediately. +// The caller must have resolved reqMsgID from a client msgs_ack through the +// sole outbound actor's tracked server-message mapping; an untrusted request +// msg_id must never call this method directly. +// +// ACK can race the asynchronous terminal Put. A pending flight records the +// event so Put resolves current waiters but does not retain a completed row. +// This follows the logical-session model: once the client has acknowledged the +// exact server message, neither its payload nor its duplicate receipt is useful. +func (c *rpcResultCache) Acknowledge(authKeyID [8]byte, sessionID, reqMsgID int64) bool { + if c == nil || reqMsgID == 0 { + return false + } + key := rpcResultCacheKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID} + s := c.shard(key) + now := s.now() + s.mu.Lock() + s.expireLocked(now) + + if elem := s.byKey[key]; elem != nil { + s.removeElement(elem) + s.mu.Unlock() + return true + } + if flight := s.pending[key]; flight != nil { + flight.acknowledged = true + s.mu.Unlock() + return true + } + s.mu.Unlock() + return false } // ObserveDependency returns a waiter for an admitted in-flight dependency, a @@ -337,7 +385,15 @@ func (c *rpcResultCache) Put(authKeyID [8]byte, sessionID, reqMsgID int64, encod func (c *rpcResultCache) putOnce(authKeyID [8]byte, sessionID, reqMsgID int64, encoded *encodedOutboundMessage) bool { key := rpcResultCacheKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID} s := c.shard(key) + logicalOutbox := c.sessions != nil + logicalReplayable := false + if logicalOutbox { + _, logicalReplayable = c.sessions.rpcResult(authKeyID, sessionID, reqMsgID) + } accountedSize := len(encoded.body) + if logicalOutbox { + accountedSize = 1 + } if accountedSize < 1 { // Every owner reserves one byte at admission. Keeping zero-length results // at the same minimum makes entry and byte capacity linearizable. @@ -361,17 +417,17 @@ func (c *rpcResultCache) putOnce(authKeyID [8]byte, sessionID, reqMsgID int64, e return true } - identity, admissionSeq, executionKnown, executionOK := rpcResultFlightMetadataLocked(s, key) + identity, admissionSeq, executionKnown, executionOK, acknowledged := rpcResultFlightMetadataLocked(s, key) var oldEntry *rpcResultCacheEntry if old != nil { oldEntry = old.Value.(*rpcResultCacheEntry) if flight == nil { - // A defensive duplicate terminal publication must never downgrade - // completed dependency/identity metadata after its flight disappeared. - identity = oldEntry.identity - admissionSeq = oldEntry.admissionSeq - executionKnown = oldEntry.executionKnown - executionOK = oldEntry.executionOK + // The first terminal publication is immutable, whether it retained an + // legacy inline body, receipt or capacity tombstone. A + // stale callback must not replace exact bytes, extend TTL, resurrect an + // ACKed body or valid receipt. + s.mu.Unlock() + return true } } @@ -396,7 +452,24 @@ func (c *rpcResultCache) putOnce(authKeyID [8]byte, sessionID, reqMsgID int64, e retainedSize := accountedSize retained := encoded capacity := false - if !reservation.resizeBytes(accountedSize) { + if acknowledged { + const receiptSize = 1 + if !reservation.resizeBytes(receiptSize) { + s.mu.Unlock() + panic("mtprotoedge: acknowledged rpc result owner lost its receipt reservation") + } + retainedSize = receiptSize + retained = nil + } else if logicalOutbox { + const receiptSize = 1 + if !reservation.resizeBytes(receiptSize) { + s.mu.Unlock() + panic("mtprotoedge: logical rpc result owner lost its receipt reservation") + } + retainedSize = receiptSize + retained = nil + capacity = !logicalReplayable + } else if !reservation.resizeBytes(accountedSize) { if flight == nil { // A direct replacement cannot discard the prior replay body. Leave it // untouched and let Put perform one cross-shard expiry reap before its @@ -436,6 +509,7 @@ func (c *rpcResultCache) putOnce(authKeyID [8]byte, sessionID, reqMsgID int64, e admissionSeq: admissionSeq, executionKnown: executionKnown, executionOK: executionOK, + acknowledged: acknowledged, capacity: capacity, reservation: reservation, } @@ -446,6 +520,11 @@ func (c *rpcResultCache) putOnce(authKeyID [8]byte, sessionID, reqMsgID int64, e // Resolve the independent in-flight entry only after either the completed // result or its replay tombstone is published under the same shard lock. subscribers, executionSubscribers, executionOK := c.completeRPCResultFlightLocked(s, key, encoded) + if acknowledged { + // ACK won before terminal publication. Current subscribers still receive + // encoded below, but no post-ACK receipt survives this critical section. + s.removeElement(elem) + } s.mu.Unlock() for _, subscriber := range subscribers { subscriber(encoded, true) @@ -461,11 +540,12 @@ func rpcResultFlightMetadataLocked(s *rpcResultCacheShard, key rpcResultCacheKey uint64, bool, bool, + bool, ) { if flight := s.pending[key]; flight != nil { - return flight.identity, flight.admissionSeq, flight.executionDone, flight.executionOK + return flight.identity, flight.admissionSeq, flight.executionDone, flight.executionOK, flight.acknowledged } - return rpcResultRequestIdentity{}, 0, false, false + return rpcResultRequestIdentity{}, 0, false, false, false } // expireCompletedResults performs the cold-path cross-shard reap used only @@ -484,6 +564,37 @@ func (c *rpcResultCache) expireCompletedResults() { } } +func (c *rpcResultCache) Close() error { + return nil +} + +func (c *rpcResultCache) CloseContext(context.Context) error { + return nil +} + +// forgetCompletedSession removes every terminal receipt for a logical session. +// SessionManager invokes it only after the session's physical producers have +// converged, so no pending owner should remain and no later terminal callback +// can recreate a receipt for the destroyed outbox. +func (c *rpcResultCache) forgetCompletedSession(authKeyID [8]byte, sessionID int64) { + if c == nil { + return + } + for i := range c.shards { + s := &c.shards[i] + s.mu.Lock() + for elem := s.order.Front(); elem != nil; { + next := elem.Next() + entry := elem.Value.(*rpcResultCacheEntry) + if entry.key.authKeyID == authKeyID && entry.key.sessionID == sessionID { + s.removeElement(elem) + } + elem = next + } + s.mu.Unlock() + } +} + func (s *rpcResultCacheShard) expireLocked(now time.Time) { for elem := s.order.Front(); elem != nil; { next := elem.Next() diff --git a/internal/mtprotoedge/rpc_result_cache_test.go b/internal/mtprotoedge/rpc_result_cache_test.go index 51d9b2af..bd72307a 100644 --- a/internal/mtprotoedge/rpc_result_cache_test.go +++ b/internal/mtprotoedge/rpc_result_cache_test.go @@ -187,8 +187,8 @@ func TestRPCResultCacheFairReservationLifecycleReturnsEveryScope(t *testing.T) { } cache.Put(auth, 1, 102, &encodedOutboundMessage{body: make([]byte, 2)}) - if got := cache.completedBytes.snapshot(); got != 3 { - t.Fatalf("replacement did not resize global bytes: %d", got) + if got := cache.completedBytes.snapshot(); got != 5 { + t.Fatalf("duplicate publication changed immutable global bytes: %d", got) } now = now.Add(rpcResultCacheTTL + time.Second) _, _ = cache.Get(auth, 1, 102) @@ -536,7 +536,7 @@ func TestRPCResultCacheDuplicatePutPreservesCompletedExecutionMetadata(t *testin cache.Put(keyID, sessionID, reqMsgID, second) replay, err := cache.Acquire(keyID, sessionID, reqMsgID) - if err != nil || replay.state != rpcResultAcquireCompleted || replay.encoded != second || + if err != nil || replay.state != rpcResultAcquireCompleted || replay.encoded != first || !replay.executionKnown || !replay.executionOK { t.Fatalf("duplicate Put metadata = %#v, err=%v", replay, err) } @@ -609,18 +609,18 @@ func TestRPCResultCacheByteBudgetReturnsOnReplaceExpiryAndCapacity(t *testing.T) cache.Put(keyID, 1, 101, &encodedOutboundMessage{body: make([]byte, 4)}) cache.Put(keyID, 1, 101, &encodedOutboundMessage{body: make([]byte, 7)}) - if got := cache.completedBytes.snapshot(); got != 7 { - t.Fatalf("completed bytes after growing replacement = %d, want 7", got) + if got := cache.completedBytes.snapshot(); got != 4 { + t.Fatalf("completed bytes after duplicate publication = %d, want 4", got) } - if usage := cache.fairBudget.sessionSnapshot(keyID, 1); usage.entries != 1 || usage.bytes != 7 { - t.Fatalf("replacement fair reservation after growth = %#v", usage) + if usage := cache.fairBudget.sessionSnapshot(keyID, 1); usage.entries != 1 || usage.bytes != 4 { + t.Fatalf("immutable fair reservation after duplicate = %#v", usage) } cache.Put(keyID, 1, 101, &encodedOutboundMessage{body: make([]byte, 2)}) - if got := cache.completedBytes.snapshot(); got != 2 { - t.Fatalf("completed bytes after shrinking replacement = %d, want 2", got) + if got := cache.completedBytes.snapshot(); got != 4 { + t.Fatalf("completed bytes after second duplicate = %d, want 4", got) } - if usage := cache.fairBudget.sessionSnapshot(keyID, 1); usage.entries != 1 || usage.bytes != 2 { - t.Fatalf("replacement fair reservation after shrink = %#v", usage) + if usage := cache.fairBudget.sessionSnapshot(keyID, 1); usage.entries != 1 || usage.bytes != 4 { + t.Fatalf("immutable fair reservation after second duplicate = %#v", usage) } now = now.Add(rpcResultCacheTTL + time.Second) @@ -712,23 +712,19 @@ func TestRPCResultCacheByteCapacityReclaimsExpiredAcrossShards(t *testing.T) { } func TestRPCResultCacheServerOptionsPropagateFairLimits(t *testing.T) { - sessionBytes := int64(maxOutboundBodyBytes) s := New(Options{ RPCGlobalMaxTasks: 6, RPCResultCacheMaxEntries: 12, - RPCResultCacheMaxBytes: sessionBytes + 2048, RPCResultCacheAuthMaxEntries: 8, - RPCResultCacheAuthMaxBytes: sessionBytes + 1024, RPCResultCacheSessionMaxEntries: 4, - RPCResultCacheSessionMaxBytes: sessionBytes, RPCResultPendingPerAuth: 3, }) - if s.rpcResults.completedEntries.max != 12 || s.rpcResults.completedBytes.max != sessionBytes+2048 { + if s.rpcResults.completedEntries.max != 12 || s.rpcResults.completedBytes.max != 12 { t.Fatalf("global option propagation = %d/%d", s.rpcResults.completedEntries.max, s.rpcResults.completedBytes.max) } budget := s.rpcResults.fairBudget - if budget.authLimit.entries != 8 || budget.authLimit.bytes != sessionBytes+1024 || - budget.sessionLimit.entries != 4 || budget.sessionLimit.bytes != sessionBytes || budget.pendingPerAuth != 3 { + if budget.authLimit.entries != 8 || budget.authLimit.bytes != 8 || + budget.sessionLimit.entries != 4 || budget.sessionLimit.bytes != 4 || budget.pendingPerAuth != 3 { t.Fatalf("fair option propagation = auth:%#v session:%#v pending:%d", budget.authLimit, budget.sessionLimit, budget.pendingPerAuth) } @@ -738,11 +734,8 @@ func TestRPCResultCacheServerOptionsFailFast(t *testing.T) { base := Options{ RPCGlobalMaxTasks: 6, RPCResultCacheMaxEntries: 12, - RPCResultCacheMaxBytes: 64 << 20, RPCResultCacheAuthMaxEntries: 8, - RPCResultCacheAuthMaxBytes: 32 << 20, RPCResultCacheSessionMaxEntries: 4, - RPCResultCacheSessionMaxBytes: 16 << 20, RPCResultPendingPerAuth: 3, } tests := []struct { @@ -750,7 +743,6 @@ func TestRPCResultCacheServerOptionsFailFast(t *testing.T) { mutate func(*Options) }{ {name: "entry hierarchy", mutate: func(o *Options) { o.RPCResultCacheAuthMaxEntries = 13 }}, - {name: "body does not fit session", mutate: func(o *Options) { o.RPCResultCacheSessionMaxBytes = maxOutboundBodyBytes - 1 }}, {name: "pending hierarchy", mutate: func(o *Options) { o.RPCResultPendingPerAuth = 7 }}, } for _, test := range tests { diff --git a/internal/mtprotoedge/rpc_result_delivery_test.go b/internal/mtprotoedge/rpc_result_delivery_test.go index aa694fcf..df9a2a22 100644 --- a/internal/mtprotoedge/rpc_result_delivery_test.go +++ b/internal/mtprotoedge/rpc_result_delivery_test.go @@ -401,8 +401,8 @@ func TestRPCResultFailureAfterIntentionalTerminalDoesNotCloseTransferLease(t *te // Session replacement publishes terminal before transferring the physical // lease. A late old-generation result sees ErrConnClosed at producer admission; - // it may publish cache-only, but must not upgrade that intentional fence into - // a physical close that makes Transfer fail. + // it publishes a metadata-only capacity tombstone, but must not upgrade that + // intentional fence into a physical close that makes Transfer fail. oldConn.beginTerminalShutdown() err = s.sendResult(context.Background(), oldConn, reqMsgID, exactTestRPCResult(&tg.Config{ThisDC: 2})) if !errors.Is(err, ErrOutboundTrackedBudget) { @@ -423,9 +423,8 @@ func TestRPCResultFailureAfterIntentionalTerminalDoesNotCloseTransferLease(t *te if tr.closed.Load() || !newConn.isPhysicalTransportCurrentOpen() { t.Fatalf("stale close after transfer: raw_closed=%v current_open=%v", tr.closed.Load(), newConn.isPhysicalTransportCurrentOpen()) } - completed, acquireErr := s.rpcResults.Acquire(key.ID, oldConn.sessionID, reqMsgID) - if acquireErr != nil || completed.state != rpcResultAcquireCompleted || completed.encoded == nil { - t.Fatalf("late result cache = %+v err=%v", completed, acquireErr) + if _, acquireErr := s.rpcResults.Acquire(key.ID, oldConn.sessionID, reqMsgID); !errors.Is(acquireErr, ErrRPCResultFlightCapacity) { + t.Fatalf("late result tombstone err=%v, want %v", acquireErr, ErrRPCResultFlightCapacity) } newConn.ForceClose() } @@ -459,9 +458,8 @@ func TestRPCResultPublishesBeforePathologicalPhysicalCloseReturns(t *testing.T) case <-time.After(time.Second): t.Fatal("pathological physical Close blocked result publication") } - completed, acquireErr := s.rpcResults.Acquire(key.ID, c.sessionID, reqMsgID) - if acquireErr != nil || completed.state != rpcResultAcquireCompleted || completed.encoded == nil { - t.Fatalf("completed result before raw Close return = %+v err=%v", completed, acquireErr) + if _, acquireErr := s.rpcResults.Acquire(key.ID, c.sessionID, reqMsgID); !errors.Is(acquireErr, ErrRPCResultFlightCapacity) { + t.Fatalf("result tombstone before raw Close return err=%v, want %v", acquireErr, ErrRPCResultFlightCapacity) } close(tr.release) if c.transportLease != nil { diff --git a/internal/mtprotoedge/rpc_result_retention_budget_test.go b/internal/mtprotoedge/rpc_result_retention_budget_test.go index b00e26a5..5aef3eab 100644 --- a/internal/mtprotoedge/rpc_result_retention_budget_test.go +++ b/internal/mtprotoedge/rpc_result_retention_budget_test.go @@ -162,7 +162,7 @@ func TestOutboundActorRetargetRequiresSecondBodyReservation(t *testing.T) { state := newOutboundState(budget) var terminalErr error var terminalBytes int64 - c.handleOutboundSend(state, outboundOp{ + op := outboundOp{ kind: outboundSend, ctx: context.Background(), msgType: proto.MessageServerResponse, @@ -174,12 +174,14 @@ func TestOutboundActorRetargetRequiresSecondBodyReservation(t *testing.T) { terminalErr = err terminalBytes = budget.snapshot() }, - }) + } + err := c.handleOutboundSend(state, op) + op.finish(outboundResult{err: err}) if !errors.Is(terminalErr, ErrOutboundTrackedBudget) { t.Fatalf("retarget terminal error = %v, want %v", terminalErr, ErrOutboundTrackedBudget) } - if terminalBytes != int64(len(encoded.body)) { - t.Fatalf("bytes visible to terminal = %d, want original body %d retained", terminalBytes, len(encoded.body)) + if terminalBytes != 0 { + t.Fatalf("bytes visible to terminal = %d, want producer reservation released", terminalBytes) } if got := tr.sends.Load(); got != 0 { t.Fatalf("retarget under one-body budget wrote %d frames, want 0", got) @@ -213,7 +215,7 @@ func TestOutboundActorRetargetTransfersOnlyReplacementToPending(t *testing.T) { state := newOutboundState(budget) var terminalErr error var terminalBytes int64 - c.handleOutboundSend(state, outboundOp{ + op := outboundOp{ kind: outboundSend, ctx: context.Background(), msgType: proto.MessageServerResponse, @@ -225,12 +227,14 @@ func TestOutboundActorRetargetTransfersOnlyReplacementToPending(t *testing.T) { terminalErr = err terminalBytes = budget.snapshot() }, - }) + } + err := c.handleOutboundSend(state, op) + op.finish(outboundResult{err: err}) if terminalErr != nil { t.Fatalf("retarget terminal error: %v", terminalErr) } - if terminalBytes != int64(2*perBody) { - t.Fatalf("bytes visible to terminal = %d, want original+replacement %d", terminalBytes, 2*perBody) + if terminalBytes != int64(perBody) { + t.Fatalf("bytes visible to terminal = %d, want only pending replacement %d", terminalBytes, perBody) } if got := budget.snapshot(); got != int64(perBody) { t.Fatalf("bytes after terminal = %d, want one pending replacement %d", got, perBody) diff --git a/internal/mtprotoedge/rpc_rewrap.go b/internal/mtprotoedge/rpc_rewrap.go index 18283cb7..e99298c4 100644 --- a/internal/mtprotoedge/rpc_rewrap.go +++ b/internal/mtprotoedge/rpc_rewrap.go @@ -510,8 +510,10 @@ const ( rpcRewrapObserverWorkers = 1 rpcRewrapObserverQueue = 64 // Queue residence, physical delivery and ordered restore share one absolute - // deadline. An admitted alias must never retain a Conn scheduler barrier for - // minutes behind older slow jobs. + // control deadline. It prevents later stages from starting and fences logical + // ownership, but cannot cancel non-cooperative filesystem, transport or restore + // work. Failure publication may wait for replay preparation to leave its + // ownership transition before releasing the Conn scheduler barrier. rpcRewrapDeliveryQueueTimeout = 5 * time.Second ) @@ -532,10 +534,10 @@ const ( rpcRewrapJobFailed ) -// rpcRewrapDeliveryControl lets an independent deadline timer retire queued, -// running and physically committed jobs. A late worker cannot enter run after -// the timer wins; a committed non-cooperative restore is fenced by fail, and -// its eventual return cannot report failure or finish the barrier a second time. +// rpcRewrapDeliveryControl lets the deadline timer independently transition +// queued, running and physically committed jobs to Failed. A late worker cannot +// enter run after the timer wins. The timer cannot cancel a physical transport +// write; a late return still cannot report failure or finish the barrier twice. type rpcRewrapDeliveryControl struct { state atomic.Uint32 timerMu sync.Mutex @@ -550,9 +552,9 @@ type rpcRewrapPhysicalOutcome struct { // waitRPCRewrapPhysicalTerminal deliberately keeps one of the four bounded // workers attached to an in-progress actor write even after the watchdog fences // the Conn. A broken transport may therefore strand at most four workers, while -// queued jobs still time out independently. If that transport later reports -// success, the worker cannot lose the logical hook merely because timeout won -// before its goroutine resumed. +// queued jobs still transition to Failed at their control deadlines. If that +// transport later reports success, the worker cannot lose the logical hook +// merely because timeout won before its goroutine resumed. func waitRPCRewrapPhysicalTerminal( c *Conn, ctx context.Context, @@ -700,12 +702,6 @@ func runRPCRewrapDeliveryJob(j rpcRewrapDeliveryJob) { if !control.transition(rpcRewrapJobPending, rpcRewrapJobRunning) { return } - if !j.deadline.IsZero() && !time.Now().Before(j.deadline) { - if control.fail() { - j.reportFailure(context.DeadlineExceeded) - } - return - } defer func() { if recovered := recover(); recovered != nil { if control.fail() { @@ -716,6 +712,12 @@ func runRPCRewrapDeliveryJob(j rpcRewrapDeliveryJob) { } control.complete() }() + if !j.deadline.IsZero() && !time.Now().Before(j.deadline) { + if control.fail() { + j.reportFailure(context.DeadlineExceeded) + } + return + } j.run(control, j.deadline) } @@ -747,9 +749,10 @@ func scheduleRPCRewrapJob( delay = 0 } timer := time.AfterFunc(delay, func() { - if job.control.timeout() { - job.reportFailure(context.DeadlineExceeded) + if !job.control.timeout() { + return } + job.reportFailure(context.DeadlineExceeded) }) job.control.installTimer(timer) select { @@ -766,6 +769,22 @@ func scheduleRPCRewrapJob( } } +func (s *Server) attachRPCRewrapReplayPreparation( + job *rpcRewrapDeliveryJob, + c *Conn, + reqMsgID int64, + method string, + encoded *encodedOutboundMessage, +) { + if job == nil || s == nil || c == nil || encoded == nil { + return + } + // Freeze scheduling metadata before the watchdog can publish the compact + // receipt. Exact wire bytes are owned only by the logical-session outbox, so + // there is no pre-send spool preparation or I/O gate. + encoded.priority = rpcResultPriority(method, encoded) +} + func scheduleRPCRewrapDeliveryJob(job rpcRewrapDeliveryJob) bool { return scheduleRPCRewrapJob(job, &rpcRewrapDeliveryOnce, &rpcRewrapDeliveryJobs, rpcRewrapDeliveryWorkers, rpcRewrapDeliveryQueue) @@ -953,6 +972,7 @@ func (a *rpcRewrapAlias) activate(s *Server) error { s.log.Debug("Retargeted RPC restore failed", zap.Error(restoreErr)) } }) + s.attachRPCRewrapReplayPreparation(&job, a.conn, a.newReqID, a.method, clone) job.fail = func(err error) { // Never enter deliveredFinalizeOnce from the timer goroutine: the worker // may already own a non-cooperative restore. Fence and release its Conn @@ -978,6 +998,7 @@ func (a *rpcRewrapAlias) activate(s *Server) error { job := s.rpcRewrapRestoreJob(a, "pending init rewrap result", func(control *rpcRewrapDeliveryControl, deadline time.Time) { s.publishRewrappedRPCResult(a.conn, a.newReqID, a.method, a.newOwner, clone, a, control, deadline) }) + s.attachRPCRewrapReplayPreparation(&job, a.conn, a.newReqID, a.method, clone) // Once this alias consumed the source candidate, expiration or panic of // the admitted worker job must still publish the immutable result under // the new msg_id. Otherwise the alias owner would remain pending forever @@ -1067,14 +1088,15 @@ func (s *Server) publishRewrappedRPCResult( alias.releaseBodyReservation() return } - priority := rpcResultPriority(method, encoded) - encoded.priority = priority - encoded.markQueued() if deadline.IsZero() { deadline = time.Now().Add(rpcRewrapDeliveryQueueTimeout) } ctx, cancel := context.WithDeadline(context.Background(), deadline) defer cancel() + // attachRPCRewrapReplayPreparation froze cache-copied scheduling metadata + // before the watchdog started. The delivery path must not mutate plain fields + // that a concurrent timeout publication can copy into the replay ledger. + encoded.markQueued() // Rewrap delivery is synchronous on this small bounded worker pool. This // makes the queue deadline cover the physical write and lets the pending // logical hook join the same per-Conn ordered restore, without touching the @@ -1093,10 +1115,10 @@ func (s *Server) publishRewrappedRPCResult( return } // Physical success outranks an already-fired watchdog. The timeout path may - // have fenced and cached a replayable clone, but it cannot revoke bytes; the + // have fenced and published a replayable receipt, but it cannot revoke bytes; the // shared once/coordinator below still completes logical state exactly once. // Run replacement metadata then the original logical hook before publishing - // the alias cache entry. Whole-finalization once also covers a watchdog racing + // the alias receipt. Whole-finalization once also covers a watchdog racing // a late physical terminal, so completed metadata cannot be overwritten. restoreParent := ctx if !outcome.owned { diff --git a/internal/mtprotoedge/rpc_rewrap_test.go b/internal/mtprotoedge/rpc_rewrap_test.go index 17a629fa..50885b0b 100644 --- a/internal/mtprotoedge/rpc_rewrap_test.go +++ b/internal/mtprotoedge/rpc_rewrap_test.go @@ -1,6 +1,7 @@ package mtprotoedge import ( + "bytes" "context" "encoding/binary" "errors" @@ -445,7 +446,7 @@ func TestInitRewrapAfterWritingReplaysWithoutBusinessExecution(t *testing.T) { if !oldOwner.HandOff() { t.Fatal("old owner handoff failed") } - s.rpcResults.Put(c.authKeyID, c.sessionID, oldReqID, encoded) + storeLogicalRPCResultForTest(t, s, c, oldReqID, encoded) deadline := time.Now().Add(2 * time.Second) var replayed *encodedOutboundMessage @@ -534,7 +535,7 @@ func TestInitRewrapAliasesExecutionAndRetargetsQueuedResult(t *testing.T) { t.Fatal("old owner handoff failed") } encoded.markDelivered() - s.rpcResults.Put(c.authKeyID, c.sessionID, oldReqID, encoded) + storeLogicalRPCResultForTest(t, s, c, oldReqID, encoded) var ( aliased *encodedOutboundMessage @@ -772,7 +773,7 @@ func TestRetargetedRPCRestoreIsOrderedAndIndependentOfGlobalHookExecutor(t *test t.Fatalf("retargeted physical req_msg_id = %d, want %d", got, newReqID) } encoded.markDelivered() - s.rpcResults.Put(c.authKeyID, c.sessionID, oldReqID, encoded) + storeLogicalRPCResultForTest(t, s, c, oldReqID, encoded) deadline := time.Now().Add(time.Second) for order.Load() != 2 && time.Now().Before(deadline) { @@ -1108,7 +1109,8 @@ func TestRPCRewrapSubscriberPanicCannotLoseClaimedLogicalHook(t *testing.T) { if pending != 0 { t.Fatalf("restore barriers after subscriber panic = %d, want 0", pending) } - if cached, ok := s.rpcResults.Get(c.authKeyID, c.sessionID, reqMsgID); !ok || cached != encoded { - t.Fatalf("completed result after subscriber panic = (%p, %v), want (%p, true)", cached, ok, encoded) + if cached, ok := s.rpcResults.Get(c.authKeyID, c.sessionID, reqMsgID); !ok || + cached.delivery != encoded.delivery || !bytes.Equal(cached.body, encoded.body) || cached.replayMsgID == 0 { + t.Fatalf("logical outbox result after subscriber panic = (%p, %v), source=%p", cached, ok, encoded) } } diff --git a/internal/mtprotoedge/rpc_test.go b/internal/mtprotoedge/rpc_test.go index 1d0ba18f..ba63003e 100644 --- a/internal/mtprotoedge/rpc_test.go +++ b/internal/mtprotoedge/rpc_test.go @@ -118,7 +118,7 @@ func TestLayerRPCGetConfigUsesExactAdmittedProfile(t *testing.T) { } } -func TestInboundRPCQueueFullReturnsFloodWait(t *testing.T) { +func TestInboundRPCQueueFullReturnsWorkerBusy(t *testing.T) { const dc = 2 handler := &blockingRPC{ started: make(chan struct{}, 1), @@ -152,8 +152,8 @@ func TestInboundRPCQueueFullReturnsFloodWait(t *testing.T) { if err := rpcErr.Decode(&bin.Buffer{Buf: result.Result}); err != nil { t.Fatalf("decode rpc_error: %v", err) } - if rpcErr.ErrorCode != 420 || rpcErr.ErrorMessage != "FLOOD_WAIT_1" { - t.Fatalf("rpc_error = %d %q, want 420 FLOOD_WAIT_1", rpcErr.ErrorCode, rpcErr.ErrorMessage) + if rpcErr.ErrorCode != rpcWorkerBusyErrorCode || rpcErr.ErrorMessage != rpcWorkerBusyErrorMessage { + t.Fatalf("rpc_error = %d %q, want %d %s", rpcErr.ErrorCode, rpcErr.ErrorMessage, rpcWorkerBusyErrorCode, rpcWorkerBusyErrorMessage) } close(handler.release) } diff --git a/internal/mtprotoedge/server.go b/internal/mtprotoedge/server.go index de0cc6a6..06a3b4ae 100644 --- a/internal/mtprotoedge/server.go +++ b/internal/mtprotoedge/server.go @@ -284,16 +284,13 @@ type Options struct { // 等于 copied body；exact charge 是 typed decode 前的保守 materialization // 上界，因此该配置不表示可并发接收 512 MiB wire body。默认 512 MiB。 RPCGlobalMaxBytes int64 - // RPCResultCache* limits bound pending ownership and completed rpc_result - // replay state across the full 331-second duplicate horizon. Every owner is - // charged simultaneously at global, raw-auth and session scopes. Defaults: - // global 262144/64 MiB, auth 32768/32 MiB, session 16384/16 MiB. + // RPCResultCache*Entries bound in-flight owners and compact completed + // receipts. Exact payload bytes are not charged here: the logical-session + // outbox owns them under OutboundTrackedGlobalMaxBytes until ACK. ACK removes + // the receipt immediately; 331 seconds is only the no-ACK safety horizon. RPCResultCacheMaxEntries int - RPCResultCacheMaxBytes int64 RPCResultCacheAuthMaxEntries int - RPCResultCacheAuthMaxBytes int64 RPCResultCacheSessionMaxEntries int - RPCResultCacheSessionMaxBytes int64 // RPCResultPendingPerAuth is an additional active-owner bound, independent // from the retained entry limits and RPCGlobalMaxTasks. Default 2048. RPCResultPendingPerAuth int @@ -395,21 +392,12 @@ func (o *Options) setDefaults() { if o.RPCResultCacheMaxEntries == 0 { o.RPCResultCacheMaxEntries = rpcResultCacheMaxEntries } - if o.RPCResultCacheMaxBytes == 0 { - o.RPCResultCacheMaxBytes = rpcResultCacheMaxBytes - } if o.RPCResultCacheAuthMaxEntries == 0 { o.RPCResultCacheAuthMaxEntries = rpcResultCacheAuthMaxEntries } - if o.RPCResultCacheAuthMaxBytes == 0 { - o.RPCResultCacheAuthMaxBytes = rpcResultCacheAuthMaxBytes - } if o.RPCResultCacheSessionMaxEntries == 0 { o.RPCResultCacheSessionMaxEntries = rpcResultCacheSessionMaxEntries } - if o.RPCResultCacheSessionMaxBytes == 0 { - o.RPCResultCacheSessionMaxBytes = rpcResultCacheSessionMaxBytes - } if o.RPCResultPendingPerAuth == 0 { o.RPCResultPendingPerAuth = rpcResultFlightMaxPendingPerAuth if o.RPCResultPendingPerAuth > o.RPCGlobalMaxTasks { @@ -457,17 +445,6 @@ func validateRPCResultCacheOptions(o Options) error { return fmt.Errorf("rpc_result cache entry hierarchy must satisfy global >= auth >= session: %d/%d/%d", o.RPCResultCacheMaxEntries, o.RPCResultCacheAuthMaxEntries, o.RPCResultCacheSessionMaxEntries) } - if o.RPCResultCacheMaxBytes < int64(maxOutboundBodyBytes) || - o.RPCResultCacheAuthMaxBytes < int64(maxOutboundBodyBytes) || - o.RPCResultCacheSessionMaxBytes < int64(maxOutboundBodyBytes) { - return fmt.Errorf("rpc_result cache byte limits must each be at least max outbound body %d: %d/%d/%d", - maxOutboundBodyBytes, o.RPCResultCacheMaxBytes, o.RPCResultCacheAuthMaxBytes, o.RPCResultCacheSessionMaxBytes) - } - if o.RPCResultCacheMaxBytes < o.RPCResultCacheAuthMaxBytes || - o.RPCResultCacheAuthMaxBytes < o.RPCResultCacheSessionMaxBytes { - return fmt.Errorf("rpc_result cache byte hierarchy must satisfy global >= auth >= session: %d/%d/%d", - o.RPCResultCacheMaxBytes, o.RPCResultCacheAuthMaxBytes, o.RPCResultCacheSessionMaxBytes) - } if o.RPCResultPendingPerAuth <= 0 || o.RPCResultPendingPerAuth > o.RPCGlobalMaxTasks || o.RPCResultPendingPerAuth > o.RPCResultCacheAuthMaxEntries { return fmt.Errorf("rpc_result per-auth pending limit %d must be positive and <= global pending %d and auth entries %d", @@ -534,7 +511,7 @@ func New(opts Options) *Server { if conns == nil { conns = NewSessionManager(opts.Logger.Named("sessions")) } - return &Server{ + server := &Server{ log: opts.Logger, codec: opts.Codec, obfuscated: opts.ObfuscatedTCP, @@ -570,16 +547,21 @@ func New(opts Options) *Server { rpcResults: newRPCResultCacheWithFairCapacity(opts.Clock.Now, rpcResultCacheCapacity{ maxPending: opts.RPCGlobalMaxTasks, maxPendingPerAuth: opts.RPCResultPendingPerAuth, - globalMaxBytes: opts.RPCResultCacheMaxBytes, + globalMaxBytes: int64(opts.RPCResultCacheMaxEntries), globalMaxEntries: opts.RPCResultCacheMaxEntries, - authMaxBytes: opts.RPCResultCacheAuthMaxBytes, + authMaxBytes: int64(opts.RPCResultCacheAuthMaxEntries), authMaxEntries: opts.RPCResultCacheAuthMaxEntries, - sessionMaxBytes: opts.RPCResultCacheSessionMaxBytes, + sessionMaxBytes: int64(opts.RPCResultCacheSessionMaxEntries), sessionMaxEntries: opts.RPCResultCacheSessionMaxEntries, + sessions: conns, }), rpcRewrap: newRPCRewrapRegistry(opts.RPCGlobalMaxTasks), admission: newAdmissionController(opts.MaxConnections, opts.MaxConnectionsPerIP, opts.MaxConcurrentHandshakes), } + conns.setLogicalSessionReleaseHook(func(key sessionKey) { + server.rpcResults.forgetCompletedSession(key.authKeyID, key.sessionID) + }) + return server } // ListenAndServe binds the public MTProto socket and immediately enters Serve. @@ -641,8 +623,16 @@ func (s *Server) buildConn(tc transport.Conn, lease *physicalTransportLease, key outboundTrackedBudget: s.outboundTrackedBudget, outboundControlTrackedBudget: s.outboundControlBudget, outboundScratchPool: s.outboundScratchPool, - rpcResultAcked: s.rpcRewrap.acknowledge, + rpcResultAcked: func(conn *Conn, reqMsgID int64) { + // The sole outbound actor invokes this only after resolving a client + // msgs_ack server msg_id through its tracked resend frame. The actor has + // already removed the sole outbox frame; now delete its receipt and + // retire any init-rewrap bookkeeping. + s.rpcResults.Acknowledge(conn.authKeyID, conn.sessionID, reqMsgID) + s.rpcRewrap.acknowledge(conn, reqMsgID) + }, } + s.conns.attachLogicalSession(c, s.outboundTrackedBudget) c.startOutbound() c.startInboundRPCScheduler(s.rpcScheduler, s.rpcInflight, s.rpcQueueSize, s.rpcTimeout) return c @@ -655,6 +645,7 @@ func (s *Server) Serve(ctx context.Context, ln net.Listener) error { // serveTCP/serveMixed 返回前会等待连接 goroutine 收敛，各 Conn 已先排空/取消任务； // 最后再停止全局池，避免关闭过程中留下无人消费但仍占预算的队列。 s.rpcScheduler.start() + defer s.conns.releaseAllLogicalSessions() defer s.rpcScheduler.stop(rpcCloseWaitTimeout) // 只在最外层 listener 包一次，确保 same-port mux 的 sniff/HTTP upgrade 也计入 // raw admission，而不是等连接已经分流后才计数。 diff --git a/internal/mtprotoedge/session_boundary_test.go b/internal/mtprotoedge/session_boundary_test.go index db0ee4b6..5084b752 100644 --- a/internal/mtprotoedge/session_boundary_test.go +++ b/internal/mtprotoedge/session_boundary_test.go @@ -156,7 +156,7 @@ func TestContainerRPCAdmissionFailureIsAtomic(t *testing.T) { if err := rpcErr.Decode(&bin.Buffer{Buf: result.Result}); err != nil { t.Fatalf("decode capacity rpc_error: %v", err) } - if rpcErr.ErrorCode != 420 || rpcErr.ErrorMessage != "FLOOD_WAIT_1" { + if rpcErr.ErrorCode != rpcWorkerBusyErrorCode || rpcErr.ErrorMessage != rpcWorkerBusyErrorMessage { t.Fatalf("capacity rpc_error = %+v", rpcErr) } delete(requestIDs, result.RequestMessageID) diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 04b31acc..0f152bea 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -171,6 +171,10 @@ func notifySessionDestroyed(observer SessionLifecycleObserver, authKeyID [8]byte type SessionManager struct { mu sync.RWMutex bySession map[sessionKey]*Conn + // logicalSessions owns MTProto resend state independently of physical Conn + // generations. It is bounded by the Server-wide tracked-body budget and a + // short offline retention window; ACK and destroy release bodies immediately. + logicalSessions map[sessionKey]*logicalSession // claims owns the provisional -> active gap. A claimant is intentionally // absent from every push/online index until its required session control frame // is on the wire and PublishActivation validates the same owner. @@ -188,6 +192,7 @@ type SessionManager struct { pending map[sessionKey][]queuedPush // updates-ready 前暂存的主动推送 flushing map[sessionKey]bool // 置位时暂存正在排空的 session；排空完成前推送继续进 pending 保序 pendingBudget *outboundTrackedBudget // 未就绪 session 暂存 encoded body 的进程级上限 + logicalSessionReleased func(sessionKey) lifecycle SessionLifecycleObserver log *zap.Logger @@ -200,6 +205,7 @@ func NewSessionManager(log *zap.Logger) *SessionManager { } return &SessionManager{ bySession: make(map[sessionKey]*Conn), + logicalSessions: make(map[sessionKey]*logicalSession), claims: make(map[sessionKey]*Conn), claimsByAuth: make(map[[8]byte]map[int64]*Conn), byAuthKey: make(map[[8]byte]map[int64]*Conn), @@ -225,6 +231,12 @@ func (m *SessionManager) SetLifecycleObserver(observer SessionLifecycleObserver) m.mu.Unlock() } +func (m *SessionManager) setLogicalSessionReleaseHook(hook func(sessionKey)) { + m.mu.Lock() + m.logicalSessionReleased = hook + m.mu.Unlock() +} + // SeedInheritedLayerForRawAuthKey supplies an auth-key-wide default to every // currently unknown active/provisional connection for rawAuthKeyID. Existing // inherited or explicit state is left untouched; only ordered invokeWithLayer @@ -634,6 +646,7 @@ func (m *SessionManager) AbortActivation(c *Conn) { m.deletePendingLocked(key) delete(m.flushing, key) } + m.markLogicalSessionOfflineLocked(key, time.Now()) owned = true } m.mu.Unlock() @@ -677,6 +690,7 @@ func (m *SessionManager) Unregister(c *Conn) { zap.Int("online", len(m.bySession)), ) } + m.markLogicalSessionOfflineLocked(key, time.Now()) m.mu.Unlock() if observer != nil { observer.SessionOffline(c.authKeyID, c.sessionID, offlineUser, lastForUser) @@ -692,6 +706,7 @@ func (m *SessionManager) DestroySessionForAuthKey(authKeyID [8]byte, sessionID i if !ok { if claim := m.claims[key]; claim != nil { m.retireClaimLocked(key, claim, true) + outbound := m.destroyLogicalSessionLocked(key) m.mu.Unlock() if !forceCloseConnBatch([]*Conn{claim}, forceCloseBatchTimeout) { m.log.Warn("Claimed session close exceeded shared deadline", @@ -699,15 +714,23 @@ func (m *SessionManager) DestroySessionForAuthKey(authKeyID [8]byte, sessionID i zap.Int64("session_id", sessionID), ) } + if outbound != nil { + m.releaseLogicalSession(key, outbound) + } notifySessionDestroyed(observer, authKeyID, sessionID) return true } m.deletePendingLocked(key) + outbound := m.destroyLogicalSessionLocked(key) m.mu.Unlock() + if outbound != nil { + m.releaseLogicalSession(key, outbound) + } notifySessionDestroyed(observer, authKeyID, sessionID) return false } offlineUser := m.retireConnLocked(c, true) + outbound := m.destroyLogicalSessionLocked(key) lastForUser := offlineUser != 0 && len(m.byUser[offlineUser]) == 0 m.log.Debug("Session destroyed", zap.String("auth_key_id", sessionKeyLog(authKeyID)), @@ -721,6 +744,9 @@ func (m *SessionManager) DestroySessionForAuthKey(authKeyID [8]byte, sessionID i zap.Int64("session_id", sessionID), ) } + if outbound != nil { + m.releaseLogicalSession(key, outbound) + } if observer != nil && offlineUser != 0 { observer.SessionOffline(authKeyID, sessionID, offlineUser, lastForUser) } @@ -815,6 +841,7 @@ func (m *SessionManager) bindAuthKeyLocked(c *Conn, key sessionKey, authKeyID [8 removeBusinessAuthKeyIndex(m.byBusinessAuthKey, oldAuthKeyID, key) } c.SetBusinessAuthKeyID(authKeyID) + m.bindLogicalSessionAuthKeyLocked(key, authKeyID) addBusinessAuthKeyIndex(m.byBusinessAuthKey, authKeyID, key, c) if changed { if oldUserID != 0 { @@ -865,6 +892,7 @@ func (m *SessionManager) CloseSessionsForBusinessAuthKey(authKeyID [8]byte) int m.mu.Lock() var conns []*Conn var events []offlineEvent + var logicalRelease []*logicalSession for key, c := range m.businessAuthKeyCandidatesLocked(authKeyID) { if !connUsesBusinessAuthKey(c, authKeyID) { continue @@ -880,6 +908,14 @@ func (m *SessionManager) CloseSessionsForBusinessAuthKey(authKeyID [8]byte) int m.retireClaimLocked(key, c, true) conns = append(conns, c) } + for key, logical := range m.logicalSessions { + if logical == nil || (key.authKeyID != authKeyID && + (!logical.businessAuthResolved || logical.businessAuthKeyID != authKeyID)) { + continue + } + delete(m.logicalSessions, key) + logicalRelease = append(logicalRelease, logical) + } observer := m.lifecycle if len(conns) > 0 { m.log.Debug("Force close sessions for revoked auth key", @@ -894,6 +930,9 @@ func (m *SessionManager) CloseSessionsForBusinessAuthKey(authKeyID [8]byte) int zap.Int("sessions", len(conns)), ) } + for _, logical := range logicalRelease { + m.releaseLogicalSession(logical.key, logical.outbound) + } if observer != nil { for _, e := range events { observer.SessionOffline(e.key.authKeyID, e.key.sessionID, e.userID, e.last) @@ -2567,6 +2606,7 @@ func (m *SessionManager) RunPendingSweeper(ctx context.Context, interval time.Du case <-ticker.C: } m.sweepStalePending() + m.sweepLogicalSessions(time.Now()) } }