merged from gramsrv upstream

internal/mtprotoedge/admission_test.go

@@ -272,7 +272,8 @@ func TestServeMixedStopsAllComponentsWhenOneReturnsCleanly(t *testing.T) {
 
 type countingAuthKeyStore struct {
 	store.AuthKeyStore
-	gets atomic.Int32
+	gets        atomic.Int32
+	revalidates atomic.Int32
 }
 
 func (s *countingAuthKeyStore) Get(ctx context.Context, id [8]byte) (store.AuthKeyData, bool, error) {
@@ -280,6 +281,11 @@ func (s *countingAuthKeyStore) Get(ctx context.Context, id [8]byte) (store.AuthK
 	return s.AuthKeyStore.Get(ctx, id)
 }
 
+func (s *countingAuthKeyStore) Revalidate(ctx context.Context, id [8]byte) (store.AuthKeyData, bool, error) {
+	s.revalidates.Add(1)
+	return s.AuthKeyStore.Revalidate(ctx, id)
+}
+
 func TestUnknownAuthKeyRespondsOnceThenCloses(t *testing.T) {
 	keys := &countingAuthKeyStore{AuthKeyStore: memory.NewAuthKeyStore()}
 	addr, _, _ := startTestServer(t, Options{AuthKeys: keys})

internal/mtprotoedge/bot_callback_e2e_test.go

@@ -230,9 +230,9 @@ func TestBotInlineKeyboardCallbackFlow(t *testing.T) {
 				return fmt.Errorf("bot getUsers(owner) = %d, want 1", len(got))
 			}
 			ownerSeen := got[0].(*tg.User)
-			markup := &tg.ReplyInlineMarkup{Rows: []tg.KeyboardButtonRow{{Buttons: []tg.KeyboardButtonClass{
-				&tg.KeyboardButtonCallback{Text: "Press", Data: callbackData},
-				&tg.KeyboardButtonURL{Text: "Site", URL: "https://example.com/x"},
+			markup := &tg.ReplyInlineMarkup{Rows: []tg.KeyboardInlineButtonRow{{Buttons: []tg.KeyboardInlineButton{
+				{Text: "Press", Type: &tg.InlineButtonTypeCallback{Data: callbackData}},
+				{Text: "Site", Type: &tg.InlineButtonTypeURL{URL: "https://example.com/x"}},
 			}}}}
 			req := &tg.MessagesSendMessageRequest{
 				Peer:     &tg.InputPeerUser{UserID: ownerSeen.ID, AccessHash: ownerSeen.AccessHash},
@@ -312,14 +312,14 @@ func TestBotInlineKeyboardCallbackFlow(t *testing.T) {
 				if !ok || len(inline.Rows) != 1 || len(inline.Rows[0].Buttons) != 2 {
 					t.Fatalf("unexpected markup shape: %#v", rm)
 				}
-				cbBtn, ok := inline.Rows[0].Buttons[0].(*tg.KeyboardButtonCallback)
+				cbBtn, ok := inline.Rows[0].Buttons[0].Type.(*tg.InlineButtonTypeCallback)
 				if !ok {
 					t.Fatalf("first button not callback: %#v", inline.Rows[0].Buttons[0])
 				}
 				if string(cbBtn.Data) != string(callbackData) {
 					t.Fatalf("callback data round-trip mismatch: got %v want %v", cbBtn.Data, callbackData)
 				}
-				if _, ok := inline.Rows[0].Buttons[1].(*tg.KeyboardButtonURL); !ok {
+				if _, ok := inline.Rows[0].Buttons[1].Type.(*tg.InlineButtonTypeURL); !ok {
 					t.Fatalf("second button not url: %#v", inline.Rows[0].Buttons[1])
 				}
 				msgID = msg.ID

internal/mtprotoedge/conn.go

@@ -118,10 +118,13 @@ type Conn struct {
 	transportClose sync.Once
 
 	rpcScheduler *inboundRPCScheduler
-	rpcCancel    context.CancelFunc
-	rpcClose     sync.Once
-	rpcMu        sync.Mutex
-	rpcWG        sync.WaitGroup
+	// rpcDeliveryHooks belongs to the owning Server. Directly constructed test
+	// Conns leave it nil and use the package test fallback.
+	rpcDeliveryHooks *rpcDeliveryHookExecutor
+	rpcCancel        context.CancelFunc
+	rpcClose         sync.Once
+	rpcMu            sync.Mutex
+	rpcWG            sync.WaitGroup
 	// rpcReservationWG 跟踪 Copy 前预算到 commit/abort 的短窗口，使 Close 返回时
 	// 全局/单连接预算都已归还或转交给明确的 queued/running task。
 	rpcReservationWG sync.WaitGroup
@@ -177,6 +180,17 @@ type Conn struct {
 	// 同步失败时保持 false，让置位短路放行、下一条 RPC 重试同步，避免
 	// 「已置位但 channel 路由缺失」的 session 静默漏收超级群推送。
 	membershipsSynced atomic.Bool
+	// updatesActivationToken/At are protected by SessionManager.mu. They make
+	// readiness activation single-flight per physical connection generation;
+	// an old delivery callback can only release the exact token it acquired.
+	updatesActivationToken uint64
+	updatesActivationAt    time.Time
+	// bootstrapProbeToken/bootstrapProbed are protected by SessionManager.mu.
+	// A delivered updates baseline performs the durable bootstrap-job probe once
+	// per physical connection generation. A failed callback releases the token;
+	// a replacement Conn starts with a fresh zero value and probes again.
+	bootstrapProbeToken uint64
+	bootstrapProbed     bool
 	// membershipGen 是本连接 channel membership 索引的修订号：任何增量修订
 	// （join/leave/kick 的 Add/Remove、身份切换/下线的整体清除）都递增。全量同步方
 	// 在读取持久成员列表前采样、落地时带回比对，检测「读取窗口内发生增量修订」的

internal/mtprotoedge/encrypted.go

@@ -237,9 +237,11 @@ func (s *Server) handleEncrypted(ctx context.Context, tc transport.Conn, cs *con
 		// BeginActivation has installed current in claimsByAuth, which is the shared
 		// linearization domain with auth-key revocation. A delete that completed before
 		// the claim is visible here as !found; a delete after this read must observe and
-		// fence the claim. This final check intentionally covers every activation path:
+		// fence the claim. Revalidate deliberately does not refresh last_used_at: the
+		// physical connection's initial Get already established the activity lease. This
+		// final check intentionally covers every activation path:
 		// first correct-salt frame, retained bad-salt provisional and session transfer.
-		fresh, found, getErr := s.authKeys.Get(ctx, current.authKeyID)
+		fresh, found, getErr := s.authKeys.Revalidate(ctx, current.authKeyID)
 		if getErr != nil {
 			return current, fmt.Errorf("revalidate activation auth key: %w", getErr)
 		}
@@ -784,6 +786,10 @@ func (s *Server) handleRPC(ctx context.Context, c *Conn, msgID int64, method str
 	}
 	dur := s.clock.Now().Sub(start)
 	s.metrics.RPCHandled(effectiveMethod, dur, dispatchErr)
+	dbSnapshot := dbStats.Snapshot()
+	if databaseMetrics, ok := s.metrics.(RPCDatabaseMetrics); ok {
+		databaseMetrics.RPCDatabase(effectiveMethod, dbSnapshot.Queries, dbSnapshot.Duration, dbSnapshot.Errors)
+	}
 	// 刷新本连接由 invokeWithLayer 证明并冻结的 exact-session layer。ok=false
 	// 表示仍无协议证据；设备/授权元数据和其它 session 都不具备回填资格。
 	if layer, ok := s.rpc.NegotiatedLayer(c.authKeyID, c.sessionID); ok {
@@ -807,7 +813,7 @@ func (s *Server) handleRPC(ctx context.Context, c *Conn, msgID int64, method str
 	if userID := c.UserID(); userID != 0 {
 		fields = append(fields, zap.Int64("user_id", userID))
 	}
-	fields = dbtrace.AppendZapFields(fields, "", dbStats.Snapshot())
+	fields = dbtrace.AppendZapFields(fields, "", dbSnapshot)
 
 	if ctxErr := ctx.Err(); ctxErr != nil {
 		// A running request owns its terminal response until Dispatch returns. If
@@ -855,7 +861,7 @@ func (s *Server) handleRPC(ctx context.Context, c *Conn, msgID int64, method str
 		}, nil)
 	}
 
-	s.log.Info("RPC handled", fields...)
+	s.log.Debug("RPC handled", fields...)
 	return s.publishRPCResult(c, msgID, effectiveMethod, owner, result, postresponse.Take(ctx))
 }
 
@@ -973,18 +979,11 @@ func (s *Server) publishRPCResult(
 	// Until enqueue transfers ownership, every exit must return the retained-byte
 	// charge. A successful transfer clears the reservation and makes this a no-op.
 	defer reserved.release()
-	priority, visible := prepareEncoded(encoded)
+	priority, _ := prepareEncoded(encoded)
 	if owner != nil && !owner.HandOff() {
 		return ErrRPCResultFlightInvalid
 	}
 
-	resultLogLevel := zap.DebugLevel
-	if visible {
-		// Keep ordinary small RPCs at debug, but make convergence and bulk/gzip
-		// delivery visible in the default service logs. These are the
-		// responses whose queueing and write latency diagnose startup Updating.
-		resultLogLevel = zap.InfoLevel
-	}
 	egressStarted := time.Now()
 	terminal := func(deliveryErr error) {
 		latency := time.Since(egressStarted)
@@ -996,7 +995,7 @@ func (s *Server) publishRPCResult(
 			encoded.markReplayable()
 			c.fenceUndeliveredRPCResult()
 			s.completeRPCResult(c, reqMsgID, encoded, true)
-			if checked := s.log.Check(resultLogLevel, "RPC result delivery fenced for replay"); checked != nil {
+			if checked := s.log.Check(zap.InfoLevel, "RPC result delivery fenced for replay"); checked != nil {
 				checked.Write(
 					zap.String("method", method), zap.Int64("req_msg_id", reqMsgID),
 					zap.Int64("delivered_req_msg_id", deliveredReqMsgID),
@@ -1008,7 +1007,7 @@ func (s *Server) publishRPCResult(
 		}
 		encoded.markDelivered()
 		s.completeRPCResult(c, reqMsgID, encoded, true)
-		if checked := s.log.Check(resultLogLevel, "RPC result delivered"); checked != nil {
+		if checked := s.log.Check(zap.DebugLevel, "RPC result delivered"); checked != nil {
 			checked.Write(
 				zap.String("method", method), zap.Int64("req_msg_id", reqMsgID),
 				zap.Int64("delivered_req_msg_id", deliveredReqMsgID),
@@ -1024,7 +1023,7 @@ func (s *Server) publishRPCResult(
 		terminal(err)
 		return err
 	}
-	if checked := s.log.Check(resultLogLevel, "RPC result admitted"); checked != nil {
+	if checked := s.log.Check(zap.DebugLevel, "RPC result admitted"); checked != nil {
 		checked.Write(
 			zap.String("method", method), zap.Int64("req_msg_id", reqMsgID),
 			zap.Int("wire_bytes", len(encoded.body)), zap.Int("inner_bytes", encoded.uncompressedBytes),
@@ -1365,7 +1364,12 @@ func (s *Server) completeRPCResult(c *Conn, reqMsgID int64, encoded *encodedOutb
 
 // sendPong 回复 mt.PingRequest / mt.PingDelayDisconnectRequest。
 func (s *Server) sendPong(ctx context.Context, c *Conn, reqMsgID, pingID int64) error {
-	return c.SendAsync(ctx, proto.MessageServerResponse, &mt.Pong{MsgID: reqMsgID, PingID: pingID})
+	// Telegram iOS keeps the account in its connection-context "updating" state
+	// until the initial actualization ping receives its matching pong. A pong is
+	// therefore a request-correlated transport barrier, not disposable keepalive
+	// noise: if it cannot be written, reconnect instead of silently stranding the
+	// client on an otherwise healthy session.
+	return c.SendRequiredControl(ctx, proto.MessageServerResponse, &mt.Pong{MsgID: reqMsgID, PingID: pingID})
 }
 
 // sendFutureSalts 回复 MTProto get_future_salts。
@@ -1388,7 +1392,10 @@ func (s *Server) sendFutureSalts(ctx context.Context, c *Conn, reqMsgID int64, n
 			Salt:       c.salt,
 		})
 	}
-	return c.SendAsync(ctx, proto.MessageServerResponse, &mt.FutureSalts{
+	// future_salts completes the client's time/salt synchronization task. If it
+	// cannot be written, fail the connection so the client reconnects instead of
+	// remaining connected in a permanent service-task state.
+	return c.SendRequiredControl(ctx, proto.MessageServerResponse, &mt.FutureSalts{
 		ReqMsgID: reqMsgID,
 		Now:      now,
 		Salts:    salts,
@@ -1401,7 +1408,7 @@ func (s *Server) sendFutureSalts(ctx context.Context, c *Conn, reqMsgID int64, n
 // （Android 收到后才调 getDifference）随之丢失。
 func (s *Server) sendNewSessionCreated(ctx context.Context, c *Conn, firstMsgID int64) error {
 	// This notification changes the client's request map and update recovery
-	// state. Unlike best-effort ack/pong traffic, it must be written successfully
+	// state. Unlike best-effort ack traffic, it must be written successfully
 	// before the corresponding RPC batch starts executing.
 	return c.SendRequiredControl(ctx, proto.MessageFromServer, &mt.NewSessionCreated{
 		FirstMsgID: firstMsgID,
@@ -1425,7 +1432,9 @@ func (s *Server) sendAck(ctx context.Context, c *Conn, ids ...int64) error {
 
 // sendMsgsStateInfo 回复 msgs_state_req/msg_resend_req。
 func (s *Server) sendMsgsStateInfo(ctx context.Context, c *Conn, reqMsgID int64, info []byte) error {
-	return c.SendAsync(ctx, proto.MessageServerResponse, &mt.MsgsStateInfo{ReqMsgID: reqMsgID, Info: info})
+	// msgs_state_info terminates the client's resend service. Do not acknowledge
+	// the request locally and then silently discard its answer from a full queue.
+	return c.SendRequiredControl(ctx, proto.MessageServerResponse, &mt.MsgsStateInfo{ReqMsgID: reqMsgID, Info: info})
 }
 
 func (s *Server) sendDestroySession(ctx context.Context, c *Conn, sessionID int64) error {

internal/mtprotoedge/exchange.go

@@ -88,7 +88,10 @@ func (s *Server) handleExchange(ctx context.Context, conn transport.Conn, first
 	}
 
 	s.metrics.HandshakeDone(s.clock.Now().Sub(start))
-	s.log.Info("Key exchange completed",
+	// Successful handshakes are already covered by the bounded latency metric.
+	// One INFO write per PFS connection turns a 10,000-account login burst into
+	// synchronous logger and filesystem pressure.
+	s.log.Debug("Key exchange completed",
 		zap.Int64("auth_key_id", res.Key.IntID()),
 		zap.Int64("server_salt", res.ServerSalt),
 		zap.Duration("dur", s.clock.Now().Sub(start)),

internal/mtprotoedge/exchange_test.go

@@ -13,7 +13,9 @@ import (
 	"testing"
 	"time"
 
+	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
+	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/gotd/log/logzap"
 	"github.com/iamxvbaba/td/bin"
@@ -43,8 +45,9 @@ func TestKeyExchange(t *testing.T) {
 	}
 
 	keys := memory.NewAuthKeyStore()
+	logCore, observedLogs := observer.New(zap.DebugLevel)
 	srv := New(Options{
-		Logger:   zaptest.NewLogger(t),
+		Logger:   zap.New(logCore),
 		DC:       dc,
 		RSAKey:   rsaKey,
 		AuthKeys: keys,
@@ -98,6 +101,10 @@ func TestKeyExchange(t *testing.T) {
 	if saved.ServerSalt != res.ServerSalt {
 		t.Fatalf("server salt mismatch: server=%d client=%d", saved.ServerSalt, res.ServerSalt)
 	}
+	completed := observedLogs.FilterMessage("Key exchange completed").All()
+	if len(completed) != 1 || completed[0].Level != zap.DebugLevel {
+		t.Fatalf("successful key exchange logs = %+v, want one Debug entry", completed)
+	}
 
 	cancel()
 	select {

internal/mtprotoedge/inbound_layer_rpc_test.go

@@ -449,7 +449,7 @@ func TestNestedExplicitLayerAdmissionErrorsAreNotDefaultFailures(t *testing.T) {
 	unsupported := exactLayerRPCBody(t, &tg.InvokeAfterMsgRequest{
 		MsgID: 1,
 		Query: &tg.InvokeWithLayerRequest{
-			Layer: 229,
+			Layer: 230,
 			Query: &tg.HelpGetConfigRequest{},
 		},
 	})
@@ -494,7 +494,7 @@ func TestNestedExplicitLayerAdmissionErrorsAreNotDefaultFailures(t *testing.T) {
 			}
 			switch test.name {
 			case "unsupported":
-				if !strings.Contains(err.Error(), "unsupported exact profile 229") {
+				if !strings.Contains(err.Error(), "unsupported exact profile 230") {
 					t.Fatalf("unsupported selector error = %v", err)
 				}
 			case "conflict":
@@ -766,7 +766,7 @@ func TestFutureExactLayerWatermarkAllowsOnlyNewerSupportedSelfHeal(t *testing.T)
 	s := New(Options{DC: 2, LayerRPC: handler})
 	authKeyID := [8]byte{0x31, 0x04}
 	const sessionID = int64(3104)
-	if _, err := handler.FreezeNegotiatedSessionLayerAt(authKeyID, sessionID, 229, 100); err != nil {
+	if _, err := handler.FreezeNegotiatedSessionLayerAt(authKeyID, sessionID, 230, 100); err != nil {
 		t.Fatal(err)
 	}
 	c := &Conn{authKeyID: authKeyID, sessionID: sessionID, metrics: NopMetrics{}}
@@ -795,7 +795,7 @@ func TestFutureExactLayerWatermarkAllowsOnlyNewerSupportedSelfHeal(t *testing.T)
 	if oldPlan.items[1].kind != inboundItemRPCAdmissionError {
 		t.Fatalf("naked non-invariant after future watermark kind=%d, want admission error", oldPlan.items[1].kind)
 	}
-	if state, rawLayer, msgID := c.layerProfileRawEvidenceState(); state.Origin != LayerProfileUnknown || rawLayer != 229 || msgID != 100 {
+	if state, rawLayer, msgID := c.layerProfileRawEvidenceState(); state.Origin != LayerProfileUnknown || rawLayer != 230 || msgID != 100 {
 		t.Fatalf("future raw watermark = %#v raw:%d msgID:%d", state, rawLayer, msgID)
 	}
 	if got := handler.publications(); len(got) != 0 {

internal/mtprotoedge/layer_profile_test.go

@@ -99,7 +99,7 @@ func TestConnSeedLayerProfile(t *testing.T) {
 }
 
 func TestConnLayerProfileRejectsUnsupported(t *testing.T) {
-	for _, profile := range []tlprofile.Profile{0, 219, 229} {
+	for _, profile := range []tlprofile.Profile{0, 219, 230} {
 		t.Run(fmt.Sprintf("layer_%d", profile), func(t *testing.T) {
 			c := &Conn{}
 			if err := c.FreezeLayerProfile(profile); !errors.Is(err, ErrLayerProfileUnsupported) {
@@ -209,7 +209,7 @@ func TestSessionManagerSeedsOnlyUnknownRawAuthKeyConnections(t *testing.T) {
 	if got := inherited.LayerProfileState(); got.Profile != tlprofile.Profile226 || got.Origin != LayerProfileInherited {
 		t.Fatalf("existing inherited connection was overwritten = %#v", got)
 	}
-	if seeded := m.SeedInheritedLayerForRawAuthKey(authKeyID, 229); seeded != 0 {
+	if seeded := m.SeedInheritedLayerForRawAuthKey(authKeyID, 230); seeded != 0 {
 		t.Fatalf("unsupported layer seeded %d connections", seeded)
 	}
 }
@@ -433,7 +433,7 @@ func TestInitialProfileSeedAvoidsPermanentKeyResolverAndPrefersPermForTemp(t *te
 		resolver := &countingInheritedLayerResolver{layer: 227, found: true}
 		s := &Server{layerRPC: resolver}
 		c := &Conn{authKeyExpiresAt: 0}
-		if err := s.seedInitialLayerProfile(context.Background(), c, 229, LayerProfileSnapshot{}); err != nil {
+		if err := s.seedInitialLayerProfile(context.Background(), c, 230, LayerProfileSnapshot{}); err != nil {
 			t.Fatal(err)
 		}
 		if resolver.calls != 0 {
@@ -445,7 +445,7 @@ func TestInitialProfileSeedAvoidsPermanentKeyResolverAndPrefersPermForTemp(t *te
 	})
 
 	t.Run("unsupported bound permanent blocks raw temp shadow", func(t *testing.T) {
-		resolver := &countingInheritedLayerResolver{layer: 229, found: true}
+		resolver := &countingInheritedLayerResolver{layer: 230, found: true}
 		s := &Server{layerRPC: resolver}
 		c := &Conn{authKeyExpiresAt: 1_900_000_000}
 		if err := s.seedInitialLayerProfile(context.Background(), c, 225, LayerProfileSnapshot{}); err != nil {
@@ -491,7 +491,7 @@ func TestInheritedLayerResolverAvailabilityUsesOnlySupportedRawTempShadow(t *tes
 		wantOrigin   LayerProfileOrigin
 	}{
 		{name: "supported raw shadow", fetchedLayer: 225, wantProfile: tlprofile.Profile225, wantOrigin: LayerProfileInherited},
-		{name: "future raw shadow stays unknown", fetchedLayer: 229, wantOrigin: LayerProfileUnknown},
+		{name: "future raw shadow stays unknown", fetchedLayer: 230, wantOrigin: LayerProfileUnknown},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			resolver := &countingInheritedLayerResolver{err: layerDurabilityUnavailableTestError{}}
@@ -553,7 +553,7 @@ func TestActivationClaimRecheckClosesTempBindLayerRace(t *testing.T) {
 	})
 
 	t.Run("unsupported permanent clears preclaim raw shadow", func(t *testing.T) {
-		resolver := &countingInheritedLayerResolver{layer: 229, found: true}
+		resolver := &countingInheritedLayerResolver{layer: 230, found: true}
 		s := &Server{layerRPC: resolver}
 		c := &Conn{authKeyID: authKeyID, sessionID: 403, authKeyExpiresAt: 1_900_000_000}
 		if err := c.SeedInheritedLayerProfile(tlprofile.Profile225); err != nil {

internal/mtprotoedge/layer_rpc_execution.go

@@ -222,6 +222,10 @@ func (s *Server) handleAdmittedLayerRPC(
 	}
 	dur := s.clock.Now().Sub(start)
 	s.metrics.RPCHandled(effectiveMethod, dur, err)
+	dbSnapshot := dbStats.Snapshot()
+	if databaseMetrics, ok := s.metrics.(RPCDatabaseMetrics); ok {
+		databaseMetrics.RPCDatabase(effectiveMethod, dbSnapshot.Queries, dbSnapshot.Duration, dbSnapshot.Errors)
+	}
 	fields := []zap.Field{
 		zap.String("method", effectiveMethod), zap.String("auth_key_id", c.authKeyHex),
 		zap.Int64("session_id", c.sessionID), zap.Int64("msg_id", msgID),
@@ -236,7 +240,7 @@ func (s *Server) handleAdmittedLayerRPC(
 	if userID := c.UserID(); userID != 0 {
 		fields = append(fields, zap.Int64("user_id", userID))
 	}
-	fields = dbtrace.AppendZapFields(fields, "", dbStats.Snapshot())
+	fields = dbtrace.AppendZapFields(fields, "", dbSnapshot)
 
 	if ctxErr := ctx.Err(); ctxErr != nil {
 		var terminal bin.Encoder
@@ -273,7 +277,7 @@ func (s *Server) handleAdmittedLayerRPC(
 			ErrorCode: 500, ErrorMessage: "INTERNAL",
 		}, nil)
 	}
-	s.log.Info("RPC handled", fields...)
+	s.log.Debug("RPC handled", fields...)
 	return s.publishAdmittedLayerRPCResult(c, msgID, effectiveMethod, owner, true, exact, postresponse.Take(ctx))
 }
 

internal/mtprotoedge/logical_session.go

@@ -49,6 +49,16 @@ func (m *SessionManager) adoptLogicalSession(c *Conn) {
 	}
 	key := connSessionKey(c)
 	m.mu.Lock()
+	// Production Conns are attached before their actor starts. A late RPC
+	// completion can race after Unregister has marked that same logical session
+	// offline; adopting the existing outbound owner must not make the physical
+	// connection live again or extend the six-minute offline horizon. Retired
+	// construction/embedded Conns must likewise not recreate a session already
+	// removed by destroy/revoke.
+	if c.isRetired() {
+		m.mu.Unlock()
+		return
+	}
 	logical := m.logicalSessions[key]
 	if logical == nil {
 		logical = &logicalSession{key: key, outbound: c.outboundState}
@@ -59,7 +69,6 @@ func (m *SessionManager) adoptLogicalSession(c *Conn) {
 		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.

internal/mtprotoedge/logical_session_test.go

@@ -138,6 +138,47 @@ func TestLogicalSessionOwnsExactResultAcrossPhysicalReconnect(t *testing.T) {
 	}
 }
 
+func TestLateCompletionAdoptionDoesNotClearOfflineRetention(t *testing.T) {
+	manager := NewSessionManager(zap.NewNop())
+	budget := newOutboundTrackedBudget(1024)
+	key := sessionKey{authKeyID: [8]byte{1, 3, 5, 7}, sessionID: 421}
+	c := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	manager.attachLogicalSession(c, budget)
+	offlineAt := time.Unix(1_800_000_000, 0)
+	manager.mu.Lock()
+	manager.markLogicalSessionOfflineLocked(key, offlineAt)
+	manager.mu.Unlock()
+
+	manager.adoptLogicalSession(c)
+	snapshot := manager.runtimeSnapshot()
+	if snapshot.logical != 1 || snapshot.offlineLogical != 1 {
+		t.Fatalf("late completion snapshot = logical:%d offline:%d, want 1/1", snapshot.logical, snapshot.offlineLogical)
+	}
+	manager.sweepLogicalSessions(offlineAt.Add(logicalSessionOfflineTTL + time.Second))
+	snapshot = manager.runtimeSnapshot()
+	if snapshot.logical != 0 || snapshot.offlineLogical != 0 {
+		t.Fatalf("post-TTL snapshot = logical:%d offline:%d, want 0/0", snapshot.logical, snapshot.offlineLogical)
+	}
+}
+
+func TestRetiredLateCompletionCannotRecreateDestroyedLogicalSession(t *testing.T) {
+	manager := NewSessionManager(zap.NewNop())
+	budget := newOutboundTrackedBudget(1024)
+	key := sessionKey{authKeyID: [8]byte{2, 4, 6, 8}, sessionID: 422}
+	c := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	manager.attachLogicalSession(c, budget)
+	c.retire()
+	manager.mu.Lock()
+	state := manager.destroyLogicalSessionLocked(key)
+	manager.mu.Unlock()
+	manager.releaseLogicalSession(key, state)
+
+	manager.adoptLogicalSession(c)
+	if snapshot := manager.runtimeSnapshot(); snapshot.logical != 0 {
+		t.Fatalf("retired late completion recreated %d logical sessions", snapshot.logical)
+	}
+}
+
 func TestLogicalSessionACKReleasesPayloadAndReceipt(t *testing.T) {
 	manager := NewSessionManager(zap.NewNop())
 	budget := newOutboundTrackedBudget(1024)

internal/mtprotoedge/metrics.go

@@ -30,6 +30,13 @@ type Metrics interface {
 	OutboundQueueWait(len, cap int)
 }
 
+// RPCDatabaseMetrics is an optional extension for request-scoped database
+// work. queries/errors are counts attributed to one RPC and duration is the
+// cumulative database time observed by the query wrapper under that request.
+type RPCDatabaseMetrics interface {
+	RPCDatabase(method string, queries int64, duration time.Duration, errors int64)
+}
+
 // RPCResultMetrics is an optional extension for the detached response pipeline.
 // Keeping it separate preserves lightweight embedders while production exporters
 // can observe preparation/compression and end-to-end delivery independently from

internal/mtprotoedge/outbound.go

@@ -1297,7 +1297,7 @@ func (c *Conn) send(ctx context.Context, t proto.MessageType, msg bin.Encoder, c
 
 func (c *Conn) SendEncoded(ctx context.Context, t proto.MessageType, encoded *encodedOutboundMessage) error {
 	if encoded != nil {
-		if err := encoded.prepareDeliveryHook(defaultRPCDeliveryHookExecutor); err != nil {
+		if err := encoded.prepareDeliveryHook(c.deliveryHookExecutor()); err != nil {
 			return err
 		}
 	}
@@ -1336,7 +1336,7 @@ func (c *Conn) enqueueEncodedDeliveryReserved(
 		return ErrConnClosed
 	}
 	if encoded != nil {
-		if err := encoded.prepareDeliveryHook(defaultRPCDeliveryHookExecutor); err != nil {
+		if err := encoded.prepareDeliveryHook(c.deliveryHookExecutor()); err != nil {
 			return err
 		}
 	}
@@ -1384,6 +1384,13 @@ func (c *Conn) enqueueEncodedDeliveryReserved(
 	return nil
 }
 
+func (c *Conn) deliveryHookExecutor() *rpcDeliveryHookExecutor {
+	if c != nil && c.rpcDeliveryHooks != nil {
+		return c.rpcDeliveryHooks
+	}
+	return defaultRPCDeliveryHookExecutor
+}
+
 func (c *Conn) sendOutbound(ctx context.Context, t proto.MessageType, msg bin.Encoder, encoded *encodedOutboundMessage, control bool) error {
 	return c.sendOutboundWithTerminal(ctx, t, msg, encoded, control, nil)
 }
@@ -1491,10 +1498,11 @@ func (c *Conn) sendOutboundWithTerminalReserved(
 	}
 }
 
-// SendAsync 入队一条 server 消息但不等待发送结果（fire-and-forget），用于读循环里的控制消息
-// （ack/pong/bad_msg/future_salts/state_info）：避免读循环被 outbound 写
-// 阻塞而连带卡死。走优先(control)队列保证不被普通 push 拖后；队列满时丢弃并记 metrics——此时
-// 连接多已严重拥塞，控制消息丢失由客户端重传 / 读写超时兜底。返回非 nil 仅表示连接已关闭。
+// SendAsync enqueues a fire-and-forget server message without waiting for the
+// physical write. It is reserved for genuinely retryable/advisory control
+// traffic such as msgs_ack: a full control queue drops the message and records
+// a metric. Request-correlated service responses and protocol corrections must
+// use SendRequiredControl so they can never be reported as sent after a drop.
 func (c *Conn) SendAsync(ctx context.Context, t proto.MessageType, msg bin.Encoder) error {
 	if c.outbound == nil || c.outboundControl == nil {
 		return ErrConnClosed

internal/mtprotoedge/outbound_required_control_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/iamxvbaba/td/bin"
+	"github.com/iamxvbaba/td/clock"
 	"github.com/iamxvbaba/td/crypto"
 	"github.com/iamxvbaba/td/mt"
 	"github.com/iamxvbaba/td/proto"
@@ -56,6 +57,75 @@ func (t *gatedRequiredControlTransport) unblock() {
 	t.closeOnce.Do(func() { close(t.release) })
 }
 
+func TestServiceTaskResponsesWaitForPhysicalWrite(t *testing.T) {
+	tests := []struct {
+		name string
+		send func(*Server, context.Context, *Conn) error
+	}{
+		{
+			name: "pong",
+			send: func(s *Server, ctx context.Context, c *Conn) error {
+				return s.sendPong(ctx, c, 11, 22)
+			},
+		},
+		{
+			name: "future_salts",
+			send: func(s *Server, ctx context.Context, c *Conn) error {
+				return s.sendFutureSalts(ctx, c, 11, 32)
+			},
+		},
+		{
+			name: "msgs_state_info",
+			send: func(s *Server, ctx context.Context, c *Conn) error {
+				return s.sendMsgsStateInfo(ctx, c, 11, []byte{4})
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			tr := newGatedRequiredControlTransport(nil)
+			c := newOutboundTestConn(t, tr, newOutboundTrackedBudget(1<<20))
+			c.outboundControlTrackedBudget = newOutboundTrackedBudget(1 << 20)
+			srv := &Server{clock: clock.System}
+
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			done := make(chan error, 1)
+			go func() {
+				done <- tc.send(srv, ctx, c)
+			}()
+
+			select {
+			case <-tr.started:
+			case <-time.After(time.Second):
+				t.Fatal("service response did not reach the physical writer")
+			}
+			select {
+			case err := <-done:
+				t.Fatalf("service response returned before physical write completed: %v", err)
+			case <-time.After(20 * time.Millisecond):
+			}
+
+			tr.unblock()
+			select {
+			case err := <-done:
+				if err != nil {
+					t.Fatalf("service response: %v", err)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("service response did not return after physical write")
+			}
+			if c.isRetired() {
+				t.Fatal("successful service response terminally closed the connection")
+			}
+			if got := tr.sends.Load(); got != 1 {
+				t.Fatalf("physical sends = %d, want 1", got)
+			}
+		})
+	}
+}
+
 func TestSendRequiredControlWaitsForPhysicalWriteAndReturnsBudget(t *testing.T) {
 	tr := newGatedRequiredControlTransport(nil)
 	controlBudget := newOutboundTrackedBudget(1 << 20)

internal/mtprotoedge/provisional_revocation_test.go

@@ -20,22 +20,21 @@ import (
 
 type activationGatedAuthKeyStore struct {
 	store.AuthKeyStore
-	gets         atomic.Int32
+	revalidates  atomic.Int32
 	finalStarted chan struct{}
 	finalRelease chan struct{}
 	startOnce    sync.Once
 }
 
-func (s *activationGatedAuthKeyStore) Get(ctx context.Context, id [8]byte) (store.AuthKeyData, bool, error) {
-	if s.gets.Add(1) == 2 {
-		s.startOnce.Do(func() { close(s.finalStarted) })
-		select {
-		case <-s.finalRelease:
-		case <-ctx.Done():
-			return store.AuthKeyData{}, false, ctx.Err()
-		}
+func (s *activationGatedAuthKeyStore) Revalidate(ctx context.Context, id [8]byte) (store.AuthKeyData, bool, error) {
+	s.revalidates.Add(1)
+	s.startOnce.Do(func() { close(s.finalStarted) })
+	select {
+	case <-s.finalRelease:
+	case <-ctx.Done():
+		return store.AuthKeyData{}, false, ctx.Err()
 	}
-	return s.AuthKeyStore.Get(ctx, id)
+	return s.AuthKeyStore.Revalidate(ctx, id)
 }
 
 func waitForManagedSessionAbsent(t *testing.T, manager *SessionManager, key sessionKey) {
@@ -79,11 +78,17 @@ func TestBadSaltStormRevalidatesStoreOnlyAtActivationBoundary(t *testing.T) {
 	if got := keys.gets.Load(); got != 1 {
 		t.Fatalf("AuthKeyStore.Get during bad-salt storm = %d, want initial lookup only", got)
 	}
+	if got := keys.revalidates.Load(); got != 0 {
+		t.Fatalf("AuthKeyStore.Revalidate during bad-salt storm = %d, want 0", got)
+	}
 
 	sendEncrypted(t, conn, cipher, auth, firstID, &tg.HelpGetConfigRequest{})
 	collectReplyFrames(t, conn, cipher, auth.AuthKey, map[uint32]int{proto.ResultTypeID: 1})
-	if got := keys.gets.Load(); got != 2 {
-		t.Fatalf("AuthKeyStore.Get after activation boundary = %d, want 2", got)
+	if got := keys.gets.Load(); got != 1 {
+		t.Fatalf("AuthKeyStore.Get after activation boundary = %d, want initial lookup only", got)
+	}
+	if got := keys.revalidates.Load(); got != 1 {
+		t.Fatalf("AuthKeyStore.Revalidate after activation boundary = %d, want 1", got)
 	}
 	waitForAtomicCalls(t, &handler.calls, 1)
 }
@@ -108,8 +113,9 @@ func TestActivationFinalAuthKeyCheckRunsAfterClaim(t *testing.T) {
 	conn, auth, cipher := dialHandshake(t, addr, dc, pub)
 	msgID := proto.NewMessageIDGen(time.Now).New(proto.MessageFromClient)
 
-	// The first Get is serveConn's decrypt lookup. The second is deliberately
-	// blocked: it must start only after BeginActivation indexed the claim.
+	// Get is serveConn's activity-bearing decrypt lookup. Revalidate is deliberately
+	// blocked: it must start only after BeginActivation indexed the claim and must
+	// not create another durable last_used_at write.
 	sendEncrypted(t, conn, cipher, auth, msgID, &tg.HelpGetConfigRequest{})
 	select {
 	case <-keys.finalStarted:

internal/mtprotoedge/rpc_delivery_hook_executor_test.go

@@ -52,6 +52,11 @@ func TestRPCDeliveryHookExecutorBoundsAdmissionWithoutBlockingDelivery(t *testin
 	if got := len(executor.slots); got != 0 {
 		t.Fatalf("executor retained %d capacity slots", got)
 	}
+	snapshot := executor.runtimeSnapshot()
+	if snapshot.workers != 1 || snapshot.capacity != 1 || snapshot.completed != 1 || snapshot.rejected != 1 ||
+		snapshot.reserved != 0 || snapshot.queued != 0 || snapshot.running != 0 || snapshot.durationSeconds <= 0 {
+		t.Fatalf("executor snapshot = %#v", snapshot)
+	}
 }
 
 func TestRPCDeliveryHookExecutorIsolatesPanicsAndContinues(t *testing.T) {
@@ -109,3 +114,58 @@ func TestEquivalentRPCDeliveryAttemptsShareExactlyOnceCoordinator(t *testing.T)
 		t.Fatalf("equivalent attempts leaked %d tickets", got)
 	}
 }
+
+func TestRPCDeliveryHookExecutorStopRejectsNewAndDrainsReserved(t *testing.T) {
+	executor := newRPCDeliveryHookExecutor(1, 2)
+	ticket, ok := executor.reserve()
+	if !ok {
+		t.Fatal("reserve ticket")
+	}
+	stopped := make(chan bool, 1)
+	go func() { stopped <- executor.stop(time.Second) }()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		executor.mu.Lock()
+		stopping := executor.stopping
+		executor.mu.Unlock()
+		if stopping {
+			break
+		}
+		time.Sleep(time.Millisecond)
+	}
+	if _, ok := executor.reserve(); ok {
+		t.Fatal("executor accepted a reservation after stop")
+	}
+	select {
+	case <-stopped:
+		t.Fatal("stop returned while a reserved ticket was still owned")
+	default:
+	}
+	ticket.release()
+	select {
+	case ok := <-stopped:
+		if !ok {
+			t.Fatal("executor did not drain before timeout")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("executor stop did not finish after ticket release")
+	}
+	snapshot := executor.runtimeSnapshot()
+	if snapshot.reserved != 0 || snapshot.queued != 0 || snapshot.running != 0 || snapshot.rejected < 1 {
+		t.Fatalf("stopped executor snapshot = %#v", snapshot)
+	}
+}
+
+func TestRPCDeliveryHookExecutorIsServerScoped(t *testing.T) {
+	first := New(Options{RPCDeliveryHookWorkers: 2, RPCDeliveryHookMaxPending: 7})
+	second := New(Options{RPCDeliveryHookWorkers: 3, RPCDeliveryHookMaxPending: 9})
+	if first.rpcDeliveryHooks == nil || second.rpcDeliveryHooks == nil || first.rpcDeliveryHooks == second.rpcDeliveryHooks {
+		t.Fatal("servers did not receive isolated delivery-hook executors")
+	}
+	firstSnapshot := first.RuntimeSnapshot()
+	secondSnapshot := second.RuntimeSnapshot()
+	if firstSnapshot.RPCDeliveryHookWorkers != 2 || firstSnapshot.RPCDeliveryHookCapacity != 7 ||
+		secondSnapshot.RPCDeliveryHookWorkers != 3 || secondSnapshot.RPCDeliveryHookCapacity != 9 {
+		t.Fatalf("server delivery-hook limits = %#v / %#v", firstSnapshot, secondSnapshot)
+	}
+}

internal/mtprotoedge/rpc_flight.go

@@ -9,7 +9,11 @@ import (
 	"github.com/iamxvbaba/td/tlprofile"
 )
 
-const rpcResultFlightDefaultMaxPending = 8192
+// A 10,000-session startup can transiently retain more than one scheduled RPC
+// per connection while downstream reads drain. Request materialization remains
+// bounded independently by RPCGlobalMaxBytes, so this count limit provides
+// queue/owner headroom without turning the scheduler into an unbounded queue.
+const rpcResultFlightDefaultMaxPending = 32768
 
 var (
 	// ErrRPCResultFlightCapacity is returned before installing a new owner when

internal/mtprotoedge/rpc_result_egress.go

@@ -8,24 +8,31 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/iamxvbaba/td/bin"
 	"github.com/iamxvbaba/td/proto"
 )
 
 const (
-	rpcResultGZIPMinBytes        = 4 << 10
-	rpcResultGZIPMaxInputBytes   = (10 << 20) - 1 // gotd client decompression hard limit.
-	rpcResultGZIPMinSavedBytes   = 1 << 10
-	rpcResultGZIPMinSavedDivisor = 12 // Require roughly 8.3% reduction.
-	rpcResultGZIPConcurrency     = 8
-	rpcDeliveryHookConcurrency   = 8
-	rpcDeliveryHookQueueSize     = 1024
+	rpcResultGZIPMinBytes            = 4 << 10
+	rpcResultGZIPMaxInputBytes       = (10 << 20) - 1 // gotd client decompression hard limit.
+	rpcResultGZIPMinSavedBytes       = 1 << 10
+	rpcResultGZIPMinSavedDivisor     = 12 // Require roughly 8.3% reduction.
+	rpcResultGZIPConcurrency         = 8
+	defaultRPCDeliveryHookWorkers    = 32
+	defaultRPCDeliveryHookMaxPending = 16_384
 )
 
 var rpcResultGZIPSlots = make(chan struct{}, rpcResultGZIPConcurrency)
 
-var defaultRPCDeliveryHookExecutor = newRPCDeliveryHookExecutor(rpcDeliveryHookConcurrency, rpcDeliveryHookQueueSize)
+// This executor is only a compatibility boundary for directly constructed
+// Conn/encoded-message tests. Production Conns always use their owning Server's
+// isolated executor.
+var defaultRPCDeliveryHookExecutor = newRPCDeliveryHookExecutor(
+	defaultRPCDeliveryHookWorkers,
+	defaultRPCDeliveryHookMaxPending,
+)
 
 // ErrRPCDeliveryHookCapacity means an RPC result with a delivery-dependent
 // transition cannot reserve reliable executor capacity. The result must not be
@@ -53,20 +60,43 @@ type rpcDeliveryHookJob struct {
 	fn     func()
 }
 
-// rpcDeliveryHookExecutor has process lifetime. Capacity bounds queued plus
-// running hooks; every physical write reserves a ticket before admission. A
-// successful writer therefore performs only one short O(1) queue append and
-// never waits for capacity or hook work. Failed writes release their ticket,
-// while the shared logical coordinator remains eligible for a later replay.
+// rpcDeliveryHookExecutor is owned by one Server. Capacity bounds reserved plus
+// queued plus running hooks; every physical write reserves a ticket before
+// admission. A successful writer therefore performs only one short O(1) queue
+// append and never waits for capacity or hook work. Failed writes release their
+// ticket, while the shared logical coordinator remains eligible for a later
+// replay.
 type rpcDeliveryHookExecutor struct {
-	slots chan struct{}
+	workers  int
+	capacity int
+	slots    chan struct{}
+	start    sync.Once
+	wg       sync.WaitGroup
 
-	mu   sync.Mutex
-	cond *sync.Cond
-	head *rpcDeliveryHookJob
-	tail *rpcDeliveryHookJob
+	mu       sync.Mutex
+	cond     *sync.Cond
+	head     *rpcDeliveryHookJob
+	tail     *rpcDeliveryHookJob
+	stopping bool
 
-	panics atomic.Uint64
+	queued        atomic.Int64
+	running       atomic.Int64
+	completed     atomic.Uint64
+	rejected      atomic.Uint64
+	panics        atomic.Uint64
+	durationNanos atomic.Uint64
+}
+
+type rpcDeliveryHookRuntimeSnapshot struct {
+	workers         int64
+	capacity        int64
+	reserved        int64
+	queued          int64
+	running         int64
+	completed       uint64
+	rejected        uint64
+	panics          uint64
+	durationSeconds float64
 }
 
 func newRPCDeliveryHookExecutor(workers, capacity int) *rpcDeliveryHookExecutor {
@@ -76,24 +106,50 @@ func newRPCDeliveryHookExecutor(workers, capacity int) *rpcDeliveryHookExecutor
 	if capacity < workers {
 		capacity = workers
 	}
-	e := &rpcDeliveryHookExecutor{slots: make(chan struct{}, capacity)}
-	e.cond = sync.NewCond(&e.mu)
-	for range workers {
-		go e.run()
+	e := &rpcDeliveryHookExecutor{
+		workers:  workers,
+		capacity: capacity,
+		slots:    make(chan struct{}, capacity),
 	}
+	e.cond = sync.NewCond(&e.mu)
 	return e
 }
 
+func (e *rpcDeliveryHookExecutor) startWorkers() {
+	if e == nil {
+		return
+	}
+	e.start.Do(func() {
+		e.mu.Lock()
+		defer e.mu.Unlock()
+		if e.stopping {
+			return
+		}
+		e.wg.Add(e.workers)
+		for range e.workers {
+			go e.run()
+		}
+	})
+}
+
 func (e *rpcDeliveryHookExecutor) reserve() (*rpcDeliveryHookTicket, bool) {
 	if e == nil {
 		return nil, false
 	}
+	e.startWorkers()
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.stopping {
+		e.rejected.Add(1)
+		return nil, false
+	}
 	select {
 	case e.slots <- struct{}{}:
 		ticket := &rpcDeliveryHookTicket{executor: e}
 		ticket.state.Store(uint32(rpcDeliveryHookTicketReserved))
 		return ticket, true
 	default:
+		e.rejected.Add(1)
 		return nil, false
 	}
 }
@@ -105,6 +161,7 @@ func (t *rpcDeliveryHookTicket) release() {
 		return
 	}
 	<-t.executor.slots
+	t.executor.signalStateChange()
 }
 
 func (t *rpcDeliveryHookTicket) submit(fn func()) bool {
@@ -127,14 +184,20 @@ func (e *rpcDeliveryHookExecutor) enqueue(job *rpcDeliveryHookJob) {
 		e.tail.next = job
 	}
 	e.tail = job
+	e.queued.Add(1)
 	e.cond.Signal()
 	e.mu.Unlock()
 }
 
 func (e *rpcDeliveryHookExecutor) run() {
+	defer e.wg.Done()
 	for {
 		e.mu.Lock()
 		for e.head == nil {
+			if e.stopping && len(e.slots) == 0 {
+				e.mu.Unlock()
+				return
+			}
 			e.cond.Wait()
 		}
 		job := e.head
@@ -143,27 +206,88 @@ func (e *rpcDeliveryHookExecutor) run() {
 			e.tail = nil
 		}
 		job.next = nil
+		e.queued.Add(-1)
+		e.running.Add(1)
 		e.mu.Unlock()
 		e.runOne(job)
 	}
 }
 
 func (e *rpcDeliveryHookExecutor) runOne(job *rpcDeliveryHookJob) {
+	started := time.Now()
 	defer func() {
 		if recovered := recover(); recovered != nil {
 			e.panics.Add(1)
 			log.Printf("mtprotoedge: rpc delivery hook panic: %v\n%s", recovered, debug.Stack())
 		}
+		e.durationNanos.Add(uint64(time.Since(started)))
+		e.completed.Add(1)
+		e.running.Add(-1)
 		if job != nil && job.ticket != nil {
 			job.ticket.state.Store(uint32(rpcDeliveryHookTicketDone))
 			<-e.slots
 		}
+		e.signalStateChange()
 	}()
 	if job != nil && job.fn != nil {
 		job.fn()
 	}
 }
 
+func (e *rpcDeliveryHookExecutor) signalStateChange() {
+	if e == nil {
+		return
+	}
+	e.mu.Lock()
+	e.cond.Broadcast()
+	e.mu.Unlock()
+}
+
+// stop rejects new reservations and lets every already-reserved ticket either
+// be released or submitted and executed. Timing out never abandons jobs: the
+// existing workers continue draining under their Server-owned executor.
+func (e *rpcDeliveryHookExecutor) stop(timeout time.Duration) bool {
+	if e == nil {
+		return true
+	}
+	e.mu.Lock()
+	e.stopping = true
+	e.cond.Broadcast()
+	e.mu.Unlock()
+	done := make(chan struct{})
+	go func() {
+		e.wg.Wait()
+		close(done)
+	}()
+	if timeout <= 0 {
+		<-done
+		return true
+	}
+	select {
+	case <-done:
+		return true
+	case <-time.After(timeout):
+		return false
+	}
+}
+
+func (e *rpcDeliveryHookExecutor) runtimeSnapshot() rpcDeliveryHookRuntimeSnapshot {
+	if e == nil {
+		return rpcDeliveryHookRuntimeSnapshot{}
+	}
+	return rpcDeliveryHookRuntimeSnapshot{
+		workers:         int64(e.workers),
+		capacity:        int64(e.capacity),
+		reserved:        int64(len(e.slots)),
+		queued:          e.queued.Load(),
+		running:         e.running.Load(),
+		completed:       e.completed.Load(),
+		rejected:        e.rejected.Load(),
+		panics:          e.panics.Load(),
+		durationSeconds: float64(e.durationNanos.Load()) / float64(time.Second),
+	}
+}
+
 // encodeAdaptiveRPCResultInner returns either the original layer-specific TL
 // object or one complete gzip_packed object. Compression is CPU bounded and is
 // retained only when it materially reduces the non-preemptible transport frame.

internal/mtprotoedge/rpc_result_egress_test.go

@@ -721,6 +721,11 @@ func TestDeliveryHookRunsOnceAfterReplayNotFailedWrite(t *testing.T) {
 	failing := &failAfterTransport{}
 	failing.failAt.Store(1)
 	oldConn := newOutboundTestConn(t, failing, newOutboundTrackedBudget(1<<20))
+	// Production attaches the logical outbox before the outbound actor starts.
+	// Keep that invariant here so a failed physical write can retire the Conn
+	// without relying on the construction-only late-adoption bridge to recreate
+	// an already-retired session.
+	s.conns.adoptLogicalSession(oldConn)
 	const reqMsgID = int64(9101)
 	claim, err := s.rpcResults.Acquire(oldConn.authKeyID, oldConn.sessionID, reqMsgID)
 	if err != nil || claim.state != rpcResultAcquireOwner {

internal/mtprotoedge/runtime_metrics.go

@@ -23,6 +23,15 @@ type RuntimeSnapshot struct {
 	InboundRPCReadyConnections     int64
 	InboundRPCMaxTasks             int64
 	InboundRPCMaxBytes             int64
+	RPCDeliveryHookWorkers         int64
+	RPCDeliveryHookCapacity        int64
+	RPCDeliveryHookReserved        int64
+	RPCDeliveryHookQueued          int64
+	RPCDeliveryHookRunning         int64
+	RPCDeliveryHookCompleted       uint64
+	RPCDeliveryHookRejected        uint64
+	RPCDeliveryHookPanics          uint64
+	RPCDeliveryHookDurationSeconds float64
 	InboundFrameBytes              int64
 	InboundFrameMaxBytes           int64
 	OutboundTrackedBytes           int64
@@ -139,21 +148,31 @@ func (s *Server) RuntimeSnapshot() RuntimeSnapshot {
 	sessions := s.conns.runtimeSnapshot()
 	admission := s.admission.runtimeSnapshot()
 	inbound := s.rpcScheduler.runtimeSnapshot()
+	deliveryHooks := s.rpcDeliveryHooks.runtimeSnapshot()
 	result := RuntimeSnapshot{
-		RawConnections:             admission.connections,
-		RawConnectionLimit:         admission.connectionLimit,
-		Handshakes:                 admission.handshakes,
-		HandshakeLimit:             admission.handshakeLimit,
-		ActiveSessions:             sessions.active,
-		ProvisionalSessions:        sessions.provisional,
-		LogicalSessions:            sessions.logical,
-		OfflineLogicalSessions:     sessions.offlineLogical,
-		LogicalOutboxFrames:        sessions.frames,
-		LogicalOutboxBytes:         sessions.bytes,
-		PendingPushBytes:           sessions.pendingBytes,
-		InboundRPCTasks:            inbound.tasks,
-		InboundRPCBytes:            inbound.bytes,
-		InboundRPCReadyConnections: inbound.ready,
+		RawConnections:                 admission.connections,
+		RawConnectionLimit:             admission.connectionLimit,
+		Handshakes:                     admission.handshakes,
+		HandshakeLimit:                 admission.handshakeLimit,
+		ActiveSessions:                 sessions.active,
+		ProvisionalSessions:            sessions.provisional,
+		LogicalSessions:                sessions.logical,
+		OfflineLogicalSessions:         sessions.offlineLogical,
+		LogicalOutboxFrames:            sessions.frames,
+		LogicalOutboxBytes:             sessions.bytes,
+		PendingPushBytes:               sessions.pendingBytes,
+		InboundRPCTasks:                inbound.tasks,
+		InboundRPCBytes:                inbound.bytes,
+		InboundRPCReadyConnections:     inbound.ready,
+		RPCDeliveryHookWorkers:         deliveryHooks.workers,
+		RPCDeliveryHookCapacity:        deliveryHooks.capacity,
+		RPCDeliveryHookReserved:        deliveryHooks.reserved,
+		RPCDeliveryHookQueued:          deliveryHooks.queued,
+		RPCDeliveryHookRunning:         deliveryHooks.running,
+		RPCDeliveryHookCompleted:       deliveryHooks.completed,
+		RPCDeliveryHookRejected:        deliveryHooks.rejected,
+		RPCDeliveryHookPanics:          deliveryHooks.panics,
+		RPCDeliveryHookDurationSeconds: deliveryHooks.durationSeconds,
 	}
 	if s.rpcScheduler != nil {
 		result.InboundRPCMaxTasks = int64(s.rpcScheduler.maxTasks)

internal/mtprotoedge/runtime_metrics_test.go

@@ -18,9 +18,13 @@ func TestRuntimeSnapshotIsNilSafeAndReportsConfiguredLimits(t *testing.T) {
 	if snapshot.RawConnectionLimit <= 0 || snapshot.HandshakeLimit <= 0 {
 		t.Fatalf("admission limits not reported: %#v", snapshot)
 	}
-	if snapshot.InboundRPCMaxTasks <= 0 || snapshot.InboundRPCMaxBytes <= 0 {
+	if snapshot.InboundRPCMaxTasks != rpcResultFlightDefaultMaxPending || snapshot.InboundRPCMaxBytes <= 0 {
 		t.Fatalf("inbound RPC limits not reported: %#v", snapshot)
 	}
+	if snapshot.RPCDeliveryHookWorkers != defaultRPCDeliveryHookWorkers ||
+		snapshot.RPCDeliveryHookCapacity != defaultRPCDeliveryHookMaxPending {
+		t.Fatalf("delivery hook limits not reported: %#v", snapshot)
+	}
 	if snapshot.InboundFrameMaxBytes <= 0 || snapshot.OutboundTrackedMaxBytes <= 0 || snapshot.OutboundWriteMaxBytes <= 0 {
 		t.Fatalf("byte limits not reported: %#v", snapshot)
 	}

internal/mtprotoedge/server.go

@@ -164,8 +164,10 @@ type LayerRPCDurableSessionProfileResolver interface {
 }
 
 // LayerRPCDurableSessionProfileAdvancer atomically advances exact-session and
-// auth-key shared-default evidence. publishShared is true only when this exact
-// observation still owns the durable shared default.
+// auth-key shared-default evidence. publishShared is true only when this call
+// established a different durable profile generation which still owns the
+// shared default; a same-generation msg_id high-water advance needs no second
+// process-local default publication.
 type LayerRPCDurableSessionProfileAdvancer interface {
 	AdvanceNegotiatedSessionLayerEvidence(
 		ctx context.Context,
@@ -290,12 +292,20 @@ type Options struct {
 	RPCTimeout time.Duration
 	// RPCGlobalWorkers 是 Server 共享 inbound RPC worker 数。默认 256。
 	RPCGlobalWorkers int
-	// RPCGlobalMaxTasks 是全进程已预留、排队和执行中的 RPC 条数上限。默认 8192。
+	// RPCGlobalMaxTasks 是全进程已预留、排队和执行中的 RPC 条数上限。默认 32768；
+	// request materialization 仍独立受 RPCGlobalMaxBytes 硬限制。
 	RPCGlobalMaxTasks int
 	// RPCGlobalMaxBytes 是上述 RPC 的进程级 memory charge 预算。legacy charge
 	// 等于 copied body；exact charge 是 typed decode 前的保守 materialization
 	// 上界，因此该配置不表示可并发接收 512 MiB wire body。默认 512 MiB。
 	RPCGlobalMaxBytes int64
+	// RPCDeliveryHookWorkers bounds concurrent post-response correctness work
+	// such as delivered-cursor commits and first-session readiness. The pending
+	// limit separately covers reserved + queued + running hooks so a 10k startup
+	// burst remains bounded without forcing socket writers to wait. Defaults are
+	// 32 workers and 16,384 pending hooks.
+	RPCDeliveryHookWorkers    int
+	RPCDeliveryHookMaxPending int
 	// RPCExecution*Entries bound in-flight owners and compact completed
 	// receipts. Payload bytes are not charged here: the logical-session
 	// outbox owns them under OutboundTrackedGlobalMaxBytes until ACK. ACK removes
@@ -415,6 +425,12 @@ func (o *Options) setDefaults() {
 	if o.RPCGlobalMaxBytes <= 0 {
 		o.RPCGlobalMaxBytes = 512 << 20
 	}
+	if o.RPCDeliveryHookWorkers <= 0 {
+		o.RPCDeliveryHookWorkers = defaultRPCDeliveryHookWorkers
+	}
+	if o.RPCDeliveryHookMaxPending <= 0 {
+		o.RPCDeliveryHookMaxPending = defaultRPCDeliveryHookMaxPending
+	}
 	if o.RPCExecutionMaxEntries == 0 {
 		o.RPCExecutionMaxEntries = rpcExecutionMaxEntries
 	}
@@ -463,6 +479,10 @@ func (o *Options) setDefaults() {
 }
 
 func validateRPCExecutionOptions(o Options) error {
+	if o.RPCDeliveryHookWorkers <= 0 || o.RPCDeliveryHookMaxPending < o.RPCDeliveryHookWorkers {
+		return fmt.Errorf("rpc delivery hook capacity must satisfy pending >= workers > 0: %d/%d",
+			o.RPCDeliveryHookMaxPending, o.RPCDeliveryHookWorkers)
+	}
 	if o.RPCExecutionMaxEntries <= 0 || o.RPCExecutionAuthMaxEntries <= 0 || o.RPCExecutionSessionMaxEntries <= 0 {
 		return fmt.Errorf("rpc execution ledger entry limits must be positive")
 	}
@@ -498,6 +518,7 @@ type Server struct {
 	rpcQueueSize             int
 	rpcTimeout               time.Duration
 	rpcScheduler             *inboundRPCScheduler
+	rpcDeliveryHooks         *rpcDeliveryHookExecutor
 	frameBudget              *inboundFrameBudget
 	outboundQueueSize        int
 	outboundControlQueueSize int
@@ -560,6 +581,7 @@ func New(opts Options) *Server {
 		rpcQueueSize:             opts.RPCQueueSize,
 		rpcTimeout:               opts.RPCTimeout,
 		rpcScheduler:             newInboundRPCScheduler(opts.RPCGlobalWorkers, opts.RPCGlobalMaxTasks, opts.RPCGlobalMaxBytes),
+		rpcDeliveryHooks:         newRPCDeliveryHookExecutor(opts.RPCDeliveryHookWorkers, opts.RPCDeliveryHookMaxPending),
 		frameBudget:              newInboundFrameBudget(opts.InboundFrameGlobalMaxBytes),
 		outboundQueueSize:        opts.OutboundQueueSize,
 		outboundControlQueueSize: opts.OutboundControlQueueSize,
@@ -659,6 +681,7 @@ func (s *Server) buildConn(tc transport.Conn, lease *physicalTransportLease, key
 		outboundTrackedBudget:        s.outboundTrackedBudget,
 		outboundControlTrackedBudget: s.outboundControlBudget,
 		outboundScratchPool:          s.outboundScratchPool,
+		rpcDeliveryHooks:             s.rpcDeliveryHooks,
 		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
@@ -679,8 +702,10 @@ func (s *Server) buildConn(tc transport.Conn, lease *physicalTransportLease, key
 func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
 	// 共享 worker 池只在 Server 真正 Serve 后允许消费，并在首条 RPC 到达时懒启动。
 	// serveTCP/serveMixed 返回前会等待连接 goroutine 收敛，各 Conn 已先排空/取消任务；
-	// 最后再停止全局池，避免关闭过程中留下无人消费但仍占预算的队列。
+	// 最后再停止共享池并排空已预留 delivery hook，避免关闭过程中
+	// 留下无人消费但仍占预算的队列。
 	s.rpcScheduler.start()
+	defer s.rpcDeliveryHooks.stop(rpcCloseWaitTimeout)
 	defer s.conns.releaseAllLogicalSessions()
 	defer s.rpcScheduler.stop(rpcCloseWaitTimeout)
 	// 只在最外层 listener 包一次，确保 same-port mux 的 sniff/HTTP upgrade 也计入

internal/mtprotoedge/session_barrier_integration_test.go

@@ -614,9 +614,6 @@ func TestCrossConnectionInflightAbortRetriesOnlyAfterOldOwnerStops(t *testing.T)
 	if got := handler.max.Load(); got != 1 {
 		t.Fatalf("old and retry handlers overlapped: max active=%d, want 1", got)
 	}
-	if got := s.rpcResults.flightLimit.snapshot(); got != 0 {
-		t.Fatalf("sequential retry leaked flight claims: %d", got)
-	}
 
 	resultCount := 0
 	deadline = time.Now().Add(2 * time.Second)
@@ -640,6 +637,11 @@ func TestCrossConnectionInflightAbortRetriesOnlyAfterOldOwnerStops(t *testing.T)
 	if resultCount != 1 {
 		t.Fatalf("sequential retry result count = %d, want 1", resultCount)
 	}
+	// Scheduler task completion only proves that result encoding/enqueue has
+	// finished. The outbound actor publishes the terminal execution receipt
+	// after the physical write, so inspect the flight only after observing that
+	// result rather than racing the actor callback.
+	waitForRPCFlightClaims(t, s.rpcResults, 0)
 	if !firstConn.isRetired() || !secondConn.isRetired() || thirdConn == nil || !thirdConn.isActive() {
 		t.Fatalf("replacement lifecycle = first:%v second:%v third:%p active:%v", firstConn.lifecycleState(), secondConn.lifecycleState(), thirdConn, thirdConn != nil && thirdConn.isActive())
 	}
@@ -656,3 +658,14 @@ func waitForAtomicCalls(t *testing.T, calls interface{ Load() int32 }, want int3
 		t.Fatalf("handler calls = %d, want %d", got, want)
 	}
 }
+
+func waitForRPCFlightClaims(t *testing.T, ledger *rpcExecutionLedger, want int64) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for ledger.flightLimit.snapshot() != want && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if got := ledger.flightLimit.snapshot(); got != want {
+		t.Fatalf("rpc flight claims = %d, want %d", got, want)
+	}
+}

internal/mtprotoedge/session_manager.go

@@ -61,6 +61,10 @@ const (
 	maxChannelSubscriptionsPerSession = 10
 	defaultChannelSubscriptionTTL     = 75 * time.Second
 	maxChannelSubscriptionTTL         = 2 * time.Minute
+	// A claim is normally released by its rpc_result delivery callback or the
+	// pending-update flush it starts. This lease only recovers the exceptional
+	// path where result encoding is replaced before the callback can be attached.
+	updatesActivationClaimTTL = time.Minute
 )
 
 // forceCloseBatchTimeout is one deadline for a whole revoke/replace/eviction batch. Conn.Close
@@ -201,7 +205,9 @@ type SessionManager struct {
 	// 去 Google 化设备）下仍能收到来电、消息等实时推送。登记后在 pushToUserWithSender
 	// 中被视为【永久就绪】，绕过 receivesUpdates 门槛直接投递，而不是排队等一个永远
 	// 不会到来的 getState。See memory: call-inactive-account-network-pause。
-	pushSessions map[[8]byte]map[int64]struct{}
+	pushSessions         map[[8]byte]map[int64]struct{}
+	updatesActivationSeq uint64
+	bootstrapProbeSeq    uint64
 
 	lifecycle SessionLifecycleObserver
 	log       *zap.Logger
@@ -821,6 +827,8 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) {
 		if old != userID {
 			m.clearSessionChannelIndexesLocked(c, key)
 			c.membershipsSynced.Store(false)
+			m.clearUpdatesActivationLocked(c)
+			m.clearBootstrapProbeLocked(c)
 			// 身份变化即丢弃暂存推送：它们属于前一个账号，flush 给新账号是跨账号泄露。
 			// 同时取消进行中的排空（runFlush 还另有 owner 校验做批内兜底）。
 			m.deletePendingLocked(key)
@@ -833,6 +841,8 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) {
 	} else {
 		m.clearSessionChannelIndexesLocked(c, key)
 		c.membershipsSynced.Store(false)
+		m.clearUpdatesActivationLocked(c)
+		m.clearBootstrapProbeLocked(c)
 		m.deletePendingLocked(key)
 		delete(m.flushing, key)
 	}
@@ -898,6 +908,8 @@ func (m *SessionManager) bindAuthKeyLocked(c *Conn, key sessionKey, authKeyID [8
 		}
 		m.clearSessionChannelIndexesLocked(c, key)
 		c.membershipsSynced.Store(false)
+		m.clearUpdatesActivationLocked(c)
+		m.clearBootstrapProbeLocked(c)
 		m.deletePendingLocked(key)
 		delete(m.flushing, key)
 		c.userID.Store(0)
@@ -1166,6 +1178,8 @@ func (m *SessionManager) UnbindAuthKey(authKeyID [8]byte) int {
 		}
 		m.clearSessionChannelIndexesLocked(c, key)
 		c.membershipsSynced.Store(false)
+		m.clearUpdatesActivationLocked(c)
+		m.clearBootstrapProbeLocked(c)
 		// 授权解除后暂存推送属于已登出的账号，不能等下一个登录者置位时 flush 出去。
 		m.deletePendingLocked(key)
 		delete(m.flushing, key)
@@ -1185,6 +1199,7 @@ func (m *SessionManager) setReceivesUpdatesLocked(c *Conn, key sessionKey, recei
 		c.receivesUpdates.Store(false)
 		m.clearSessionChannelIndexesLocked(c, key)
 		c.membershipsSynced.Store(false)
+		m.clearUpdatesActivationLocked(c)
 		// 取消进行中的排空激活：runFlush 在置位前会复查该标志，标志已删则放弃置位，
 		// 避免把刚置 false 的开关翻回 true。
 		delete(m.flushing, key)
@@ -1198,11 +1213,15 @@ func (m *SessionManager) setReceivesUpdatesLocked(c *Conn, key sessionKey, recei
 		c.receivesUpdates.Store(false)
 		m.clearSessionChannelIndexesLocked(c, key)
 		c.membershipsSynced.Store(false)
+		m.clearUpdatesActivationLocked(c)
 		delete(m.flushing, key)
 		return 0, false
 	}
 	if c.receivesUpdates.Load() || m.flushing[key] {
 		// 已就绪，或已有排空协程在跑（完成时会自行取走新增暂存并置位）。
+		if c.receivesUpdates.Load() {
+			m.clearUpdatesActivationLocked(c)
+		}
 		return 0, false
 	}
 	if len(m.pending[key]) == 0 {
@@ -1232,6 +1251,7 @@ func (m *SessionManager) runFlush(c *Conn, key sessionKey, owner int64, attempt
 			// 排空期间发生登出/换号：剩余暂存属于旧账号，丢弃且不得发给新账号。
 			m.deletePendingLocked(key)
 			delete(m.flushing, key)
+			m.clearUpdatesActivationLocked(c)
 			m.mu.Unlock()
 			return
 		}
@@ -1239,6 +1259,7 @@ func (m *SessionManager) runFlush(c *Conn, key sessionKey, owner int64, attempt
 		if len(batch) == 0 {
 			c.receivesUpdates.Store(true)
 			delete(m.flushing, key)
+			m.clearUpdatesActivationLocked(c)
 			m.mu.Unlock()
 			return
 		}
@@ -1250,6 +1271,7 @@ func (m *SessionManager) runFlush(c *Conn, key sessionKey, owner int64, attempt
 				m.mu.Lock()
 				m.deletePendingLocked(key)
 				delete(m.flushing, key)
+				m.clearUpdatesActivationLocked(c)
 				m.mu.Unlock()
 				releaseQueuedPushes(batch[i:])
 				return
@@ -1296,6 +1318,7 @@ func (m *SessionManager) runFlush(c *Conn, key sessionKey, owner int64, attempt
 				if c.userID.Load() != owner {
 					m.deletePendingLocked(key)
 					delete(m.flushing, key)
+					m.clearUpdatesActivationLocked(c)
 				}
 				m.mu.Unlock()
 				releaseQueuedPushes(batch[i:])
@@ -1316,6 +1339,7 @@ func (m *SessionManager) runFlush(c *Conn, key sessionKey, owner int64, attempt
 				c.receivesUpdates.Store(true)
 				m.deletePendingLocked(key)
 				delete(m.flushing, key)
+				m.clearUpdatesActivationLocked(c)
 				m.mu.Unlock()
 				m.log.Debug("Flush gave up after retries; activated with getDifference fallback",
 					zap.String("auth_key_id", sessionKeyLog(key.authKeyID)),
@@ -1356,6 +1380,119 @@ func (m *SessionManager) ReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID
 	return hasProfile && c.receivesUpdates.Load() && c.membershipsSynced.Load()
 }
 
+// BeginSessionUpdatesActivation claims the readiness transition for the
+// current physical connection. Ordinary startup RPCs race here before they
+// register delivery hooks, so at most one of them can enqueue the expensive
+// channel-membership synchronization. Cursor commits remain request-owned.
+func (m *SessionManager) BeginSessionUpdatesActivation(authKeyID [8]byte, sessionID int64) (uint64, bool) {
+	if m == nil {
+		return 0, false
+	}
+	key := sessionKey{authKeyID: authKeyID, sessionID: sessionID}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	c := m.bySession[key]
+	if c == nil || c.isRetired() {
+		return 0, false
+	}
+	if _, hasProfile := c.LayerProfile(); hasProfile && c.receivesUpdates.Load() && c.membershipsSynced.Load() {
+		return 0, false
+	}
+	now := time.Now()
+	if c.now != nil {
+		now = c.now()
+	}
+	if c.updatesActivationToken != 0 {
+		// A pending FIFO flush owns the activation until it reaches a terminal
+		// outcome. Never lease-steal while that ordered delivery is in progress.
+		if m.flushing[key] || now.Sub(c.updatesActivationAt) < updatesActivationClaimTTL {
+			return 0, false
+		}
+	}
+	m.updatesActivationSeq++
+	if m.updatesActivationSeq == 0 {
+		m.updatesActivationSeq++
+	}
+	c.updatesActivationToken = m.updatesActivationSeq
+	c.updatesActivationAt = now
+	return c.updatesActivationToken, true
+}
+
+// EndSessionUpdatesActivation releases only the token owned by the caller and
+// only on the same current physical Conn. If SetReceivesUpdates started an
+// ordered pending flush, that flush retains and releases the claim itself.
+func (m *SessionManager) EndSessionUpdatesActivation(authKeyID [8]byte, sessionID int64, token uint64) {
+	if m == nil || token == 0 {
+		return
+	}
+	key := sessionKey{authKeyID: authKeyID, sessionID: sessionID}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	c := m.bySession[key]
+	if c == nil || c.updatesActivationToken != token || m.flushing[key] {
+		return
+	}
+	m.clearUpdatesActivationLocked(c)
+}
+
+// BeginSessionBootstrapProbe claims the first durable bootstrap-job lookup for
+// the current physical connection generation. Unlike updates activation, this
+// is completed only by a delivered getState/getDifference baseline.
+func (m *SessionManager) BeginSessionBootstrapProbe(authKeyID [8]byte, sessionID int64) (uint64, bool) {
+	if m == nil {
+		return 0, false
+	}
+	key := sessionKey{authKeyID: authKeyID, sessionID: sessionID}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	c := m.bySession[key]
+	if c == nil || c.isRetired() || c.bootstrapProbed || c.bootstrapProbeToken != 0 {
+		return 0, false
+	}
+	m.bootstrapProbeSeq++
+	if m.bootstrapProbeSeq == 0 {
+		m.bootstrapProbeSeq++
+	}
+	c.bootstrapProbeToken = m.bootstrapProbeSeq
+	return c.bootstrapProbeToken, true
+}
+
+// EndSessionBootstrapProbe completes or releases only the token on the same
+// current Conn. A delayed callback from a replaced connection cannot mutate the
+// replacement's one-shot state.
+func (m *SessionManager) EndSessionBootstrapProbe(authKeyID [8]byte, sessionID int64, token uint64, success bool) {
+	if m == nil || token == 0 {
+		return
+	}
+	key := sessionKey{authKeyID: authKeyID, sessionID: sessionID}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	c := m.bySession[key]
+	if c == nil || c.bootstrapProbeToken != token {
+		return
+	}
+	c.bootstrapProbeToken = 0
+	if success {
+		c.bootstrapProbed = true
+	}
+}
+
+func (m *SessionManager) clearUpdatesActivationLocked(c *Conn) {
+	if c == nil {
+		return
+	}
+	c.updatesActivationToken = 0
+	c.updatesActivationAt = time.Time{}
+}
+
+func (m *SessionManager) clearBootstrapProbeLocked(c *Conn) {
+	if c == nil {
+		return
+	}
+	c.bootstrapProbeToken = 0
+	c.bootstrapProbed = false
+}
+
 // SetReceivesUpdatesForAuthKey 标记指定 raw auth_key_id + session_id 是否接收主动 updates。
 func (m *SessionManager) SetReceivesUpdatesForAuthKey(authKeyID [8]byte, sessionID int64, receives bool) {
 	m.mu.Lock()
@@ -1470,11 +1607,32 @@ func (m *SessionManager) PushToUserAuthKeyTransient(ctx context.Context, userID
 	return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, 0, t, msg, timeout)
 }
 
-func (m *SessionManager) PushToUserAuthKeyTransientAtLeastLayer(ctx context.Context, userID int64, businessAuthKeyID [8]byte, minLayer int, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
-	return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, minLayer, t, msg, timeout)
+func (m *SessionManager) PushToUserAuthKeyTransientCompatible(ctx context.Context, userID int64, businessAuthKeyID [8]byte, semantic tlprofile.SemanticID, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
+	return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, semantic, t, msg, timeout)
 }
 
-func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, userID int64, businessAuthKeyID [8]byte, minLayer int, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
+// PushToUserExceptBusinessAuthKey 把 update 投给账号其它设备，精确排除同一 permanent
+// business auth key 下的所有 raw/temp/PFS 连接。密聊 accept 用它让输掉竞态的设备收敛为
+// discarded，同时保证获胜设备的其它连接不会误删刚建立的密聊。
+func (m *SessionManager) PushToUserExceptBusinessAuthKey(ctx context.Context, userID int64, excludeBusinessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
+	getUpdates := onceLayerUpdatesFanout(ctx, msg)
+	return m.pushToUserWithSender(ctx, userID, nil, 0, &excludeBusinessAuthKeyID, 0, t, getUpdates, false, func(c *Conn) error {
+		if c.outbound == nil || c.outboundControl == nil {
+			return ErrConnClosed
+		}
+		updates, err := getUpdates()
+		if err != nil {
+			return err
+		}
+		encoded, err := updates.prepareForConn(ctx, c)
+		if err != nil {
+			return err
+		}
+		return c.SendBestEffortEncoded(ctx, t, encoded, timeout)
+	})
+}
+
+func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, userID int64, businessAuthKeyID [8]byte, semantic tlprofile.SemanticID, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
 	if ctx != nil && ctx.Err() != nil {
 		return 0, ctx.Err()
 	}
@@ -1497,7 +1655,7 @@ func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, us
 		defer cancel()
 	}
 	getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
-	return m.pushToBusinessAuthKey(ctx, userID, businessAuthKeyID, minLayer, func(c *Conn) error {
+	return m.pushToBusinessAuthKey(ctx, userID, businessAuthKeyID, semantic, func(c *Conn) error {
 		if c.outbound == nil || c.outboundControl == nil {
 			return ErrConnClosed
 		}
@@ -1520,7 +1678,7 @@ func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, us
 	})
 }
 
-func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64, businessAuthKeyID [8]byte, minLayer int, send func(*Conn) error) (int, error) {
+func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64, businessAuthKeyID [8]byte, semantic tlprofile.SemanticID, send func(*Conn) error) (int, error) {
 	m.mu.Lock()
 	candidates := m.businessAuthKeyCandidatesLocked(businessAuthKeyID)
 	conns := make([]*Conn, 0, len(candidates))
@@ -1532,7 +1690,7 @@ func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64
 			// 未就绪：密聊消息靠 getDifference 补，typing 直接丢——都不进 pending。
 			continue
 		}
-		if !sessionSupportsMinimumLayer(c, minLayer) {
+		if !sessionSupportsSemantic(c, semantic) {
 			continue
 		}
 		conns = append(conns, c)
@@ -1579,7 +1737,7 @@ func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64
 
 func (m *SessionManager) pushToUser(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass) (int, error) {
 	getUpdates := onceLayerUpdatesFanout(ctx, msg)
-	return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, 0, t, getUpdates, true, func(c *Conn) error {
+	return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, nil, 0, t, getUpdates, true, func(c *Conn) error {
 		if c.outbound == nil || c.outboundControl == nil {
 			return ErrConnClosed
 		}
@@ -1602,7 +1760,7 @@ func (m *SessionManager) pushToUser(ctx context.Context, userID int64, excludeAu
 // 「durable 兜底」丢弃。走 best-effort 发送，不阻塞调用方。
 func (m *SessionManager) PushToUserTransientExceptAuthKeySession(ctx context.Context, userID int64, excludeAuthKeyID [8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
 	getUpdates := onceLayerUpdatesFanout(ctx, msg)
-	return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, 0, t, getUpdates, false, func(c *Conn) error {
+	return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, nil, 0, t, getUpdates, false, func(c *Conn) error {
 		if c.outbound == nil || c.outboundControl == nil {
 			return ErrConnClosed
 		}
@@ -1618,9 +1776,9 @@ func (m *SessionManager) PushToUserTransientExceptAuthKeySession(ctx context.Con
 	})
 }
 
-func (m *SessionManager) PushToUserTransientAtLeastLayer(ctx context.Context, userID int64, minLayer int, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
+func (m *SessionManager) PushToUserTransientCompatible(ctx context.Context, userID int64, semantic tlprofile.SemanticID, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
 	getUpdates := onceLayerUpdatesFanout(ctx, msg)
-	return m.pushToUserWithSender(ctx, userID, nil, 0, minLayer, t, getUpdates, false, func(c *Conn) error {
+	return m.pushToUserWithSender(ctx, userID, nil, 0, nil, semantic, t, getUpdates, false, func(c *Conn) error {
 		if c.outbound == nil || c.outboundControl == nil {
 			return ErrConnClosed
 		}
@@ -1641,10 +1799,6 @@ func (m *SessionManager) PushToUserExceptAuthKeySessionBestEffort(ctx context.Co
 }
 
 func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
-	return m.pushToUserBestEffortAtLeastLayer(ctx, userID, excludeAuthKeyID, excludeSessionID, 0, t, msg, timeout)
-}
-
-func (m *SessionManager) pushToUserBestEffortAtLeastLayer(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, minLayer int, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
 	if ctx != nil && ctx.Err() != nil {
 		return 0, ctx.Err()
 	}
@@ -1670,7 +1824,7 @@ func (m *SessionManager) pushToUserBestEffortAtLeastLayer(ctx context.Context, u
 		defer cancel()
 	}
 	getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
-	return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, minLayer, t, getUpdates, true, func(c *Conn) error {
+	return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, nil, 0, t, getUpdates, true, func(c *Conn) error {
 		if c.outbound == nil || c.outboundControl == nil {
 			return ErrConnClosed
 		}
@@ -1717,7 +1871,7 @@ func onceLayerUpdatesFanout(ctx context.Context, msg tg.UpdatesClass) func() (*l
 	}
 }
 
-func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, minLayer int, t proto.MessageType, getUpdates func() (*layerUpdatesFanout, error), queueWhenNotReady bool, send func(*Conn) error) (int, error) {
+func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64, excludeAuthKeyID *[8]byte, excludeSessionID int64, excludeBusinessAuthKeyID *[8]byte, semantic tlprofile.SemanticID, t proto.MessageType, getUpdates func() (*layerUpdatesFanout, error), queueWhenNotReady bool, send func(*Conn) error) (int, error) {
 	// push fan-out 是连接层最热路径之一：debug 日志的字段构造（含 auth_key hex 格式化）
 	// 在关闭 debug 时也会求值，先查级别一次、按需记日志。
 	debug := m.log.Core().Enabled(zapcore.DebugLevel)
@@ -1733,11 +1887,11 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
 	skipped := 0
 	needQueue := false
 	for key, c := range m.byUser[userID] {
-		if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) {
+		if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) || shouldExcludeBusinessAuthKey(c, excludeBusinessAuthKeyID) {
 			excluded++
 			continue
 		}
-		if !sessionSupportsMinimumLayer(c, minLayer) {
+		if !sessionSupportsSemantic(c, semantic) {
 			skipped++
 			continue
 		}
@@ -1768,11 +1922,11 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
 		m.mu.Lock()
 		total = len(m.byUser[userID])
 		for key, c := range m.byUser[userID] {
-			if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) {
+			if shouldExcludeSession(c, excludeAuthKeyID, excludeSessionID) || shouldExcludeBusinessAuthKey(c, excludeBusinessAuthKeyID) {
 				excluded++
 				continue
 			}
-			if !sessionSupportsMinimumLayer(c, minLayer) {
+			if !sessionSupportsSemantic(c, semantic) {
 				skipped++
 				continue
 			}
@@ -1825,6 +1979,9 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
 		if c.userID.Load() != userID {
 			continue
 		}
+		if shouldExcludeBusinessAuthKey(c, excludeBusinessAuthKeyID) {
+			continue
+		}
 		if err := send(c); err != nil {
 			if isOutboundStaleLayerEpoch(err) {
 				// Do not classify profile correction as slow-consumer evidence.
@@ -2330,6 +2487,7 @@ func (m *SessionManager) removeLocked(c *Conn, dropPending bool) int64 {
 		removeUserIndex(m.byUser, uid, key)
 	}
 	m.clearSessionChannelIndexesLocked(c, key)
+	m.clearUpdatesActivationLocked(c)
 	if dropPending {
 		m.deletePendingLocked(key)
 	}
@@ -2783,15 +2941,26 @@ func shouldExcludeSession(c *Conn, excludeAuthKeyID *[8]byte, excludeSessionID i
 	return c.authKeyID == *excludeAuthKeyID
 }
 
-func sessionSupportsMinimumLayer(c *Conn, minLayer int) bool {
-	if minLayer <= 0 {
+func shouldExcludeBusinessAuthKey(c *Conn, excludeBusinessAuthKeyID *[8]byte) bool {
+	if c == nil || excludeBusinessAuthKeyID == nil || *excludeBusinessAuthKeyID == ([8]byte{}) {
+		return false
+	}
+	return connUsesBusinessAuthKey(c, *excludeBusinessAuthKeyID)
+}
+
+func sessionSupportsSemantic(c *Conn, semantic tlprofile.SemanticID) bool {
+	if semantic == 0 {
 		return true
 	}
 	if c == nil {
 		return false
 	}
 	state := c.LayerProfileState()
-	return state.Origin != LayerProfileUnknown && int(state.Profile) >= minLayer
+	if state.Origin == LayerProfileUnknown {
+		return false
+	}
+	_, ok := tlprofile.WireID(state.Profile, semantic)
+	return ok
 }
 
 func sessionKeyLog(id [8]byte) string {

internal/mtprotoedge/session_manager_test.go

@@ -1071,6 +1071,110 @@ func TestSessionManagerWithholdsUpdatesReadinessUntilExactProfile(t *testing.T)
 	}
 }
 
+func TestSessionUpdatesActivationIsSingleFlightAndGenerationFenced(t *testing.T) {
+	sm := NewSessionManager(zaptest.NewLogger(t))
+	key := sessionKey{authKeyID: [8]byte{0x51}, sessionID: 5100}
+	old := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	if err := sm.Register(old); err != nil {
+		t.Fatal(err)
+	}
+	oldToken, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID)
+	if !ok || oldToken == 0 {
+		t.Fatal("first physical generation did not acquire activation")
+	}
+	if token, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatalf("concurrent activation acquired token %d", token)
+	}
+	sm.EndSessionUpdatesActivation(key.authKeyID, key.sessionID, oldToken+1)
+	if token, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatal("wrong-token release cleared active claim")
+	}
+
+	replacement := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	if err := sm.Register(replacement); err != nil {
+		t.Fatal(err)
+	}
+	newToken, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID)
+	if !ok || newToken == 0 || newToken == oldToken {
+		t.Fatalf("replacement activation token = %d, old %d", newToken, oldToken)
+	}
+	sm.EndSessionUpdatesActivation(key.authKeyID, key.sessionID, oldToken)
+	if token, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatal("old generation callback cleared replacement claim")
+	}
+	sm.EndSessionUpdatesActivation(key.authKeyID, key.sessionID, newToken)
+	if token, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID); !ok || token == 0 {
+		t.Fatal("matching replacement token did not release claim")
+	}
+}
+
+func TestSessionUpdatesActivationLeaseRecoversAbandonedClaim(t *testing.T) {
+	sm := NewSessionManager(zaptest.NewLogger(t))
+	key := sessionKey{authKeyID: [8]byte{0x52}, sessionID: 5200}
+	now := time.Unix(1700000000, 0)
+	c := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID, now: func() time.Time { return now }}
+	if err := sm.Register(c); err != nil {
+		t.Fatal(err)
+	}
+	first, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID)
+	if !ok {
+		t.Fatal("first activation claim rejected")
+	}
+	now = now.Add(updatesActivationClaimTTL - time.Second)
+	if token, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatal("live activation lease was stolen")
+	}
+	now = now.Add(2 * time.Second)
+	second, ok := sm.BeginSessionUpdatesActivation(key.authKeyID, key.sessionID)
+	if !ok || second == 0 || second == first {
+		t.Fatalf("expired activation lease was not replaced: first=%d second=%d", first, second)
+	}
+}
+
+func TestSessionBootstrapProbeIsOneShotRetryableAndGenerationFenced(t *testing.T) {
+	sm := NewSessionManager(zaptest.NewLogger(t))
+	key := sessionKey{authKeyID: [8]byte{0x53}, sessionID: 5300}
+	old := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	if err := sm.Register(old); err != nil {
+		t.Fatal(err)
+	}
+	failedToken, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID)
+	if !ok || failedToken == 0 {
+		t.Fatal("first bootstrap probe was not claimed")
+	}
+	if token, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatalf("concurrent bootstrap probe acquired token %d", token)
+	}
+	sm.EndSessionBootstrapProbe(key.authKeyID, key.sessionID, failedToken, false)
+	oldToken, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID)
+	if !ok || oldToken == 0 || oldToken == failedToken {
+		t.Fatalf("failed probe did not become retryable: failed=%d retry=%d", failedToken, oldToken)
+	}
+
+	replacement := &Conn{authKeyID: key.authKeyID, sessionID: key.sessionID}
+	if err := sm.Register(replacement); err != nil {
+		t.Fatal(err)
+	}
+	newToken, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID)
+	if !ok || newToken == 0 || newToken == oldToken {
+		t.Fatalf("replacement bootstrap token = %d, old %d", newToken, oldToken)
+	}
+	sm.EndSessionBootstrapProbe(key.authKeyID, key.sessionID, oldToken, true)
+	if token, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatal("old generation callback cleared replacement probe")
+	}
+	sm.EndSessionBootstrapProbe(key.authKeyID, key.sessionID, newToken, true)
+	if token, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID); ok || token != 0 {
+		t.Fatal("successful bootstrap probe was not one-shot")
+	}
+
+	sm.BindUserForAuthKey(key.authKeyID, key.sessionID, 100)
+	sm.BindUserForAuthKey(key.authKeyID, key.sessionID, 200)
+	if token, ok := sm.BeginSessionBootstrapProbe(key.authKeyID, key.sessionID); !ok || token == 0 {
+		t.Fatal("user identity change did not reset bootstrap probe")
+	}
+}
+
 func TestPendingPushBodiesUseGlobalByteBudgetAndReleaseOnDrop(t *testing.T) {
 	sm := NewSessionManager(zaptest.NewLogger(t))
 	msg := &tg.UpdateShort{Update: &tg.UpdateLoginToken{}, Date: 1700000000}

internal/mtprotoedge/transient_push_test.go

@@ -54,10 +54,10 @@ func TestPushTransientSkipsNotReadySession(t *testing.T) {
 	}
 }
 
-// Layer-228-only transient constructors must be filtered before encoding. A
-// Layer 227 or unknown session is skipped without disconnecting it or queuing
-// an unreplayable update, while the ready Layer 228 session receives it.
-func TestPushTransientAtLeastLayerSkipsOldAndUnknownProfiles(t *testing.T) {
+// Constructor compatibility comes from generated profile metadata, not a
+// hard-coded minimum layer. Old/unknown sessions are skipped without encoding,
+// disconnecting or queuing, while every generated compatible profile receives.
+func TestPushTransientCompatibleSkipsUnavailableAndUnknownProfiles(t *testing.T) {
 	sm := NewSessionManager(zaptest.NewLogger(t))
 	const userID = int64(101)
 	makeConn := func(sessionID int64, profile tlprofile.Profile, known bool) *Conn {
@@ -80,25 +80,26 @@ func TestPushTransientAtLeastLayerSkipsOldAndUnknownProfiles(t *testing.T) {
 		return c
 	}
 	old := makeConn(1, tlprofile.Profile227, true)
-	current := makeConn(2, tlprofile.Profile228, true)
+	introduced := makeConn(2, tlprofile.Profile228, true)
 	unknown := makeConn(3, 0, false)
+	newer := makeConn(4, tlprofile.Profile229, true)
 
 	message := tg.EphemeralMessage{
 		ID: 7, FromID: &tg.PeerUser{UserID: 2001}, PeerID: &tg.PeerChannel{ChannelID: 3001},
 		ReceiverID: userID, Date: 1_900_000_000, Message: "private",
 	}
 	updates := &tg.Updates{Updates: []tg.UpdateClass{&tg.UpdateNewEphemeralMessage{Message: message}}, Date: 1_900_000_000}
-	sent, err := sm.PushToUserTransientAtLeastLayer(context.Background(), userID, 228, proto.MessageFromServer, updates, time.Second)
-	if err != nil || sent != 1 {
+	sent, err := sm.PushToUserTransientCompatible(context.Background(), userID, tlprofile.SemanticTypeUpdateNewEphemeralMessage, proto.MessageFromServer, updates, time.Second)
+	if err != nil || sent != 2 {
 		t.Fatalf("sent=%d err=%v", sent, err)
 	}
-	if len(old.outbound) != 0 || len(unknown.outbound) != 0 || len(current.outbound) != 1 {
-		t.Fatalf("queues old=%d unknown=%d current=%d", len(old.outbound), len(unknown.outbound), len(current.outbound))
+	if len(old.outbound) != 0 || len(unknown.outbound) != 0 || len(introduced.outbound) != 1 || len(newer.outbound) != 1 {
+		t.Fatalf("queues old=%d unknown=%d introduced=%d newer=%d", len(old.outbound), len(unknown.outbound), len(introduced.outbound), len(newer.outbound))
 	}
 	if old.isRetired() || unknown.isRetired() {
 		t.Fatal("unsupported transient update retired an old/unknown session")
 	}
-	for _, c := range []*Conn{old, current, unknown} {
+	for _, c := range []*Conn{old, introduced, unknown, newer} {
 		sm.mu.RLock()
 		pending := len(sm.pending[connSessionKey(c)])
 		sm.mu.RUnlock()