refactor(mtproto): sync redesign RPC replay ownership

.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

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,

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

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 不新增环境变量。代码硬限制为：每个

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 {

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")

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 一次。

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)
 }
 

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

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,

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:

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})

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)
 	}

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)
+}

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)
+	}
+}

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
+	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.
+		// 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) {
+		acked = state.ack(op.ids)
+	case outboundQueryState:
+		result.info = state.stateInfo(op.ids)
+	case outboundResend:
+		result.info, result.err = c.handleOutboundResend(state, op.ctx, op.ids)
+	case outboundResendByRequest:
+		result.resent, result.err = c.handleOutboundResendByRequest(state, op.ctx, op.reqMsgID)
+	default:
+		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)
 		}
 	}
-	case outboundQueryState:
-		op.finish(outboundResult{info: state.stateInfo(op.ids)})
-	case outboundResend:
-		info, err := c.handleOutboundResend(state, op.ctx, op.ids)
-		op.finish(outboundResult{info: info, err: err})
-	case outboundResendByRequest:
-		resent, err := c.handleOutboundResendByRequest(state, op.ctx, op.reqMsgID)
-		op.finish(outboundResult{resent: resent, err: err})
-	default:
-		op.finish(outboundResult{err: fmt.Errorf("unknown outbound op %d", op.kind)})
-	}
+	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),
+		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

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,
+	}
+}

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,

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
@@ -47,6 +42,9 @@ type rpcResultCacheKey struct {
 
 type rpcResultCacheEntry struct {
 	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()

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 {

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 {

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)

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 {

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)
 	}
 }

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)
 }

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，而不是等连接已经分流后才计数。

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)

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())
 	}
 }