package mtprotoedge

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"io"
	"sync"
	"sync/atomic"
	"testing"
	"time"
	"unsafe"

	"github.com/iamxvbaba/td/bin"
	"github.com/iamxvbaba/td/crypto"
	"github.com/iamxvbaba/td/mt"
	"github.com/iamxvbaba/td/proto"
	"github.com/iamxvbaba/td/tg"
	"github.com/iamxvbaba/td/tlprofile"
	"github.com/iamxvbaba/td/transport"
)

type staticRPCReplaySource struct {
	inner []byte
}

func (s *staticRPCReplaySource) EncodeInner(_ context.Context, out *bin.Buffer) error {
	out.Put(s.inner)
	return nil
}

func (*staticRPCReplaySource) RetainedBytes() int { return 128 }

type failAfterTransport struct {
	failAt atomic.Int32
	sends  atomic.Int32
	stored atomic.Int32
	closes atomic.Int32
	mu     sync.Mutex
	last   []byte
}

type acknowledgementCaptureMetrics struct {
	NopMetrics
	count      atomic.Int64
	bytes      atomic.Int64
	retainedNS atomic.Int64
	rpcResult  atomic.Bool
}

func (m *acknowledgementCaptureMetrics) LogicalOutboxAcknowledged(bytes int, retainedFor time.Duration, rpcResult bool) {
	m.count.Add(1)
	m.bytes.Add(int64(bytes))
	m.retainedNS.Store(int64(retainedFor))
	m.rpcResult.Store(rpcResult)
}

func TestRPCResultReplayAttemptHooksArePhysicalConnectionLocal(t *testing.T) {
	const reqMsgID = int64(771)
	base := &encodedOutboundMessage{
		body:     make([]byte, 12),
		typeID:   proto.ResultTypeID,
		reqMsgID: reqMsgID,
		delivery: newRPCResultDelivery(reqMsgID),
	}
	var logical, firstAttempt, secondAttempt atomic.Int32
	base.setDeliveryHook(func() { logical.Add(1) })
	first, err := cloneRPCResultForRequest(base, reqMsgID, false)
	if err != nil {
		t.Fatal(err)
	}
	second, err := cloneRPCResultForRequest(base, reqMsgID, false)
	if err != nil {
		t.Fatal(err)
	}
	first.setAttemptDeliveryHook(func() { firstAttempt.Add(1) })
	second.setAttemptDeliveryHook(func() { secondAttempt.Add(1) })
	first.markDelivered()
	deadline := time.Now().Add(time.Second)
	for (logical.Load() != 1 || firstAttempt.Load() != 1) && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if logical.Load() != 1 || firstAttempt.Load() != 1 || secondAttempt.Load() != 0 {
		t.Fatalf("first delivery hooks = logical:%d first:%d second:%d", logical.Load(), firstAttempt.Load(), secondAttempt.Load())
	}
	second.markDelivered()
	deadline = time.Now().Add(time.Second)
	for secondAttempt.Load() != 1 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if logical.Load() != 1 || firstAttempt.Load() != 1 || secondAttempt.Load() != 1 {
		t.Fatalf("second delivery hooks = logical:%d first:%d second:%d", logical.Load(), firstAttempt.Load(), secondAttempt.Load())
	}
}

type blockingOutboundTransport struct {
	started chan struct{}
	release chan struct{}
	once    sync.Once
	sends   atomic.Int32
}

type blockingEncodeProbe struct {
	started chan struct{}
	release <-chan struct{}
	active  atomic.Int32
	max     atomic.Int32
}

func (e *blockingEncodeProbe) Encode(b *bin.Buffer) error {
	active := e.active.Add(1)
	for {
		max := e.max.Load()
		if active <= max || e.max.CompareAndSwap(max, active) {
			break
		}
	}
	e.started <- struct{}{}
	<-e.release
	e.active.Add(-1)
	b.PutID(tg.UpdatesTooLongTypeID)
	return nil
}

func newBlockingOutboundTransport() *blockingOutboundTransport {
	return &blockingOutboundTransport{started: make(chan struct{}), release: make(chan struct{})}
}

func TestOutboundEncodingHasProcessWideConcurrencyBudget(t *testing.T) {
	const extra = 8
	total := defaultOutboundEncodeConcurrency + extra
	release := make(chan struct{})
	probe := &blockingEncodeProbe{
		started: make(chan struct{}, total),
		release: release,
	}
	errs := make(chan error, total)
	for range total {
		go func() {
			_, err := encodeOutboundMessage(probe)
			errs <- err
		}()
	}

	for range defaultOutboundEncodeConcurrency {
		select {
		case <-probe.started:
		case <-time.After(time.Second):
			t.Fatal("encode workers did not fill concurrency budget")
		}
	}
	select {
	case <-probe.started:
		t.Fatalf("more than %d outbound encodes ran concurrently", defaultOutboundEncodeConcurrency)
	case <-time.After(50 * time.Millisecond):
	}

	close(release)
	for range total {
		if err := <-errs; err != nil {
			t.Fatalf("encode: %v", err)
		}
	}
	if got := probe.max.Load(); got != defaultOutboundEncodeConcurrency {
		t.Fatalf("peak concurrent encodes = %d, want %d", got, defaultOutboundEncodeConcurrency)
	}
}

func TestConnectionCloseDoesNotWaitForRunningEncoder(t *testing.T) {
	release := make(chan struct{})
	probe := &blockingEncodeProbe{started: make(chan struct{}, 1), release: release}
	c := &Conn{metrics: NopMetrics{}}
	c.startOutbound()
	sendDone := make(chan error, 1)
	go func() {
		sendDone <- c.Send(context.Background(), proto.MessageFromServer, probe)
	}()
	select {
	case <-probe.started:
	case <-time.After(time.Second):
		t.Fatal("encoder did not start")
	}

	closeDone := make(chan struct{})
	go func() {
		c.Close()
		close(closeDone)
	}()
	select {
	case <-closeDone:
	case <-time.After(time.Second):
		t.Fatal("Conn.Close waited for external Encoder")
	}
	close(release)
	select {
	case err := <-sendDone:
		if !errors.Is(err, ErrConnClosed) {
			t.Fatalf("send after concurrent close = %v, want ErrConnClosed", err)
		}
	case <-time.After(time.Second):
		t.Fatal("send did not return after encoder release")
	}
}

func TestOutboundControlVectorsUseGlobalByteBudget(t *testing.T) {
	budget := newOutboundTrackedBudget(16)
	c := &Conn{outboundControlTrackedBudget: budget}
	op, err := c.newOutboundVectorOp(outboundAck, []int64{1, 2})
	if err != nil {
		t.Fatalf("reserve first vector: %v", err)
	}
	if got := budget.snapshot(); got != 16 {
		t.Fatalf("tracked bytes after reserve = %d, want 16", got)
	}
	if _, err := c.newOutboundVectorOp(outboundResend, []int64{3}); !errors.Is(err, ErrOutboundTrackedBudget) {
		t.Fatalf("reserve over budget error = %v, want %v", err, ErrOutboundTrackedBudget)
	}
	op.releaseReservation(budget)
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("tracked bytes after release = %d, want 0", got)
	}
}

func TestEncodedControlFramesUseIndependentBudgetForQueuedAndPendingLifetime(t *testing.T) {
	bodyBudget := newOutboundTrackedBudget(4)
	controlBudget := newOutboundTrackedBudget(256)
	tr := &failAfterTransport{}
	c := newOutboundTestConn(t, tr, bodyBudget)
	c.outboundControlTrackedBudget = controlBudget
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	// One content frame fills the ordinary body budget and remains pending.
	if err := c.SendEncoded(ctx, proto.MessageFromServer, exactTestUpdatesTooLong(t, c)); err != nil {
		t.Fatalf("fill body budget: %v", err)
	}
	first, err := crypto.NewClientCipher(rand.Reader).DecryptFromBuffer(c.key, &bin.Buffer{Buf: tr.lastFrame()})
	if err != nil {
		t.Fatalf("decrypt ordinary frame: %v", err)
	}
	if got := bodyBudget.snapshot(); got != 4 {
		t.Fatalf("body budget = %d, want saturated 4", got)
	}

	created := &mt.NewSessionCreated{FirstMsgID: 1, UniqueID: 2, ServerSalt: 3}
	encodedCreated, err := encodeOutboundMessageWithoutSlot(created)
	if err != nil {
		t.Fatalf("encode new_session_created: %v", err)
	}
	if err := c.SendAsync(ctx, proto.MessageFromServer, created); err != nil {
		t.Fatalf("new_session_created under saturated body budget: %v", err)
	}
	deadline := time.Now().Add(time.Second)
	for tr.stored.Load() < 2 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if got := tr.stored.Load(); got != 2 {
		t.Fatalf("completed physical sends = %d, want 2", got)
	}
	second, err := crypto.NewClientCipher(rand.Reader).DecryptFromBuffer(c.key, &bin.Buffer{Buf: tr.lastFrame()})
	if err != nil {
		t.Fatalf("decrypt control frame: %v", err)
	}
	if got := bodyBudget.snapshot(); got != 4 {
		t.Fatalf("body budget after control send = %d, want unchanged 4", got)
	}
	if got := controlBudget.snapshot(); got != int64(len(encodedCreated.body)) {
		t.Fatalf("control pending budget = %d, want new_session_created body %d", got, len(encodedCreated.body))
	}
	select {
	case <-c.outboundDone:
		t.Fatal("ordinary body pressure closed a healthy connection")
	default:
	}

	// Pong is non-pending, but must also remain admissible and return its control bytes after write.
	if err := c.SendAsync(ctx, proto.MessageServerResponse, &mt.Pong{MsgID: 4, PingID: 5}); err != nil {
		t.Fatalf("pong under saturated body budget: %v", err)
	}
	deadline = time.Now().Add(time.Second)
	for tr.stored.Load() < 3 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if got := tr.stored.Load(); got != 3 {
		t.Fatalf("completed physical sends after pong = %d, want 3", got)
	}
	if got := controlBudget.snapshot(); got != int64(len(encodedCreated.body)) {
		t.Fatalf("control budget after non-pending pong = %d, want pending %d", got, len(encodedCreated.body))
	}

	c.AckServerMessages([]int64{first.MessageID, second.MessageID})
	deadline = time.Now().Add(time.Second)
	for (bodyBudget.snapshot() != 0 || controlBudget.snapshot() != 0) && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if got := bodyBudget.snapshot(); got != 0 {
		t.Fatalf("body budget after ACK = %d, want 0", got)
	}
	if got := controlBudget.snapshot(); got != 0 {
		t.Fatalf("control budget after ACK = %d, want 0", got)
	}
}

func TestOutboundScratchPoolBoundsConcurrentWireCopies(t *testing.T) {
	pool := newOutboundScratchPool(300 + 2*maxCompatPacketOverhead)
	first, err := pool.acquire(context.Background(), nil, 100) // Full wire+codec+obfuscation budget.
	if err != nil {
		t.Fatalf("acquire first scratch: %v", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	if _, err := pool.acquire(ctx, nil, 100); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("second concurrent acquire = %v, want deadline backpressure", err)
	}
	pool.release(first)
	if got := pool.snapshot(); got != 100 {
		t.Fatalf("idle retained scratch = %d, want 100", got)
	}
	second, err := pool.acquire(context.Background(), nil, 100)
	if err != nil {
		t.Fatalf("reuse retained scratch: %v", err)
	}
	pool.release(second)
	if got := pool.snapshot(); got != 100 {
		t.Fatalf("scratch after reuse = %d, want one bounded idle buffer", got)
	}
}

func TestOutboundScratchPoolAccountsRetainedCodecScratch(t *testing.T) {
	pool := newOutboundScratchPool(300 + 2*maxCompatPacketOverhead)
	scratch, err := pool.acquire(context.Background(), nil, 100)
	if err != nil {
		t.Fatalf("acquire scratch: %v", err)
	}
	scratch.codec = make([]byte, 0, 80)
	pool.release(scratch)
	if got := pool.snapshot(); got != 180 {
		t.Fatalf("retained wire+codec scratch = %d, want 180", got)
	}

	reused, err := pool.acquire(context.Background(), nil, 100)
	if err != nil {
		t.Fatalf("reuse scratch: %v", err)
	}
	if cap(reused.codec) != 80 {
		t.Fatalf("reused codec scratch capacity = %d, want 80", cap(reused.codec))
	}
	pool.release(reused)
}

func TestOutboundScratchAdmissionUsesWriteTimeoutWithoutClosingHealthyConnection(t *testing.T) {
	wireBytes := encryptedOutboundWireLen(4)
	pool := newOutboundScratchPool(int64(wireBytes*3 + 2*maxCompatPacketOverhead))
	blocker, err := pool.acquire(context.Background(), nil, wireBytes)
	if err != nil {
		t.Fatalf("occupy shared scratch budget: %v", err)
	}

	tr := &failAfterTransport{}
	c := newOutboundTestConn(t, tr, newOutboundTrackedBudget(1<<20))
	c.outboundScratchPool = pool
	c.writeTimeout = 25 * time.Millisecond

	start := time.Now()
	err = c.SendEncoded(context.Background(), proto.MessageFromServer, exactTestUpdatesTooLong(t, c))
	elapsed := time.Since(start)
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("scratch admission err = %v, want deadline exceeded", err)
	}
	if elapsed > 250*time.Millisecond {
		t.Fatalf("scratch admission waited %v, want writeTimeout-bounded wait", elapsed)
	}
	if got := tr.sends.Load(); got != 0 {
		t.Fatalf("writer called %d times without scratch, want 0", got)
	}
	if c.isRetired() {
		t.Fatal("scratch admission timeout terminally closed a healthy connection")
	}
	select {
	case <-c.outboundDone:
		t.Fatal("outbound actor exited after pre-write scratch timeout")
	default:
	}

	pool.release(blocker)
	c.writeTimeout = time.Second
	if err := c.SendEncoded(context.Background(), proto.MessageFromServer, exactTestUpdatesTooLong(t, c)); err != nil {
		t.Fatalf("send after scratch capacity returned: %v", err)
	}
	if got := tr.sends.Load(); got != 1 {
		t.Fatalf("writer calls after recovery = %d, want 1", got)
	}
}

func (t *blockingOutboundTransport) Send(context.Context, *bin.Buffer) error {
	if t.sends.Add(1) == 1 {
		close(t.started)
	}
	<-t.release
	return io.ErrClosedPipe
}

func (t *blockingOutboundTransport) Recv(context.Context, *bin.Buffer) error { return io.EOF }
func (t *blockingOutboundTransport) Close() error {
	t.once.Do(func() { close(t.release) })
	return nil
}

func (t *failAfterTransport) Send(_ context.Context, b *bin.Buffer) error {
	n := t.sends.Add(1)
	if failAt := t.failAt.Load(); failAt > 0 && n >= failAt {
		return io.ErrClosedPipe
	}
	t.mu.Lock()
	t.last = append(t.last[:0], b.Raw()...)
	t.mu.Unlock()
	t.stored.Add(1)
	return nil
}

func (t *failAfterTransport) Recv(context.Context, *bin.Buffer) error { return io.EOF }
func (t *failAfterTransport) Close() error {
	t.closes.Add(1)
	return nil
}

func (t *failAfterTransport) lastFrame() []byte {
	t.mu.Lock()
	defer t.mu.Unlock()
	return append([]byte(nil), t.last...)
}

func newOutboundFailureTestConn(t *testing.T, tr transport.Conn) *Conn {
	return newOutboundTestConn(t, tr, nil)
}

func newOutboundTestConn(t *testing.T, tr transport.Conn, budget *outboundTrackedBudget) *Conn {
	t.Helper()
	var key crypto.Key
	if _, err := rand.Read(key[:]); err != nil {
		t.Fatalf("rand key: %v", err)
	}
	c := &Conn{
		transport:             tr,
		writer:                tr,
		cipher:                crypto.NewServerCipher(rand.Reader),
		msgID:                 proto.NewMessageIDGen(time.Now),
		writeTimeout:          time.Second,
		metrics:               NopMetrics{},
		key:                   key.WithID(),
		salt:                  123,
		sessionID:             456,
		outboundTrackedBudget: budget,
	}
	legacyCanonicalTestConn(t, c)
	c.startOutbound()
	t.Cleanup(c.Close)
	return c
}

func TestOutboundQueueBackingUsesSmallConfigurableBounds(t *testing.T) {
	if slot, wide := unsafe.Sizeof((*outboundOp)(nil)), unsafe.Sizeof(outboundOp{}); slot >= wide {
		t.Fatalf("indirect queue slot = %d bytes, wide outbound op = %d bytes", slot, wide)
	}
	t.Run("defaults", func(t *testing.T) {
		c := &Conn{metrics: NopMetrics{}}
		c.startOutbound()
		defer c.Close()
		if got := cap(c.outbound) + cap(c.outboundCritical) + cap(c.outboundBulk); got != defaultOutboundQueueSize {
			t.Fatalf("ordinary lane total cap = %d, want %d", got, defaultOutboundQueueSize)
		}
		if got := cap(c.outboundControl); got != defaultOutboundControlQueueSize {
			t.Fatalf("control queue cap = %d, want %d", got, defaultOutboundControlQueueSize)
		}
	})

	t.Run("configured", func(t *testing.T) {
		c := &Conn{
			metrics:                  NopMetrics{},
			outboundQueueSize:        7,
			outboundControlQueueSize: 3,
		}
		c.startOutbound()
		defer c.Close()
		if got := cap(c.outbound) + cap(c.outboundCritical) + cap(c.outboundBulk); got != 7 {
			t.Fatalf("ordinary lane total cap = %d, want 7", got)
		}
		if got := cap(c.outboundControl); got != 3 {
			t.Fatalf("control queue cap = %d, want 3", got)
		}
	})
}

func TestOutboundOpPoolClearsReferencesAndBoundsIdle(t *testing.T) {
	pool := newOutboundOpPool(1)
	op := pool.acquire()
	op.ctx = context.Background()
	op.msg = &mt.PingRequest{PingID: 1}
	op.encoded = &encodedOutboundMessage{body: []byte("payload")}
	op.ids = []int64{1, 2, 3}
	op.done = make(chan outboundResult, 1)
	op.terminal = func(error) {}
	pool.release(op)

	reused := pool.acquire()
	if reused != op {
		t.Fatal("idle outbound op was not reused")
	}
	if reused.ctx != nil || reused.msg != nil || reused.encoded != nil || reused.ids != nil || reused.done != nil || reused.terminal != nil {
		t.Fatalf("reused outbound op retained references: %+v", reused)
	}
	pool.release(reused)
	pool.release(&outboundOp{})
	if got := len(pool.idle); got != 1 {
		t.Fatalf("idle outbound op count = %d, want bounded 1", got)
	}
}

func BenchmarkOutboundOpPool(b *testing.B) {
	pool := newOutboundOpPool(1)
	b.ReportAllocs()
	for b.Loop() {
		op := pool.acquire()
		op.kind = outboundSend
		pool.release(op)
	}
}

func TestOutboundAckHistoryUsesStableCircularBacking(t *testing.T) {
	state := newOutboundState(newOutboundTrackedBudget(1 << 20))
	for id := int64(1); id <= maxTrackedAckedMsgIDs; id++ {
		state.markAcked(id)
	}
	if len(state.ackOrder) != maxTrackedAckedMsgIDs || len(state.acked) != maxTrackedAckedMsgIDs {
		t.Fatalf("initial ack history = order:%d map:%d", len(state.ackOrder), len(state.acked))
	}
	backing := &state.ackOrder[0]
	for id := int64(maxTrackedAckedMsgIDs + 1); id <= 4*maxTrackedAckedMsgIDs; id++ {
		state.markAcked(id)
	}
	if &state.ackOrder[0] != backing {
		t.Fatal("full ack history replaced its circular backing")
	}
	if len(state.ackOrder) != maxTrackedAckedMsgIDs || len(state.acked) != maxTrackedAckedMsgIDs {
		t.Fatalf("steady ack history = order:%d map:%d", len(state.ackOrder), len(state.acked))
	}
	if state.isKnown(1) || !state.isKnown(4*maxTrackedAckedMsgIDs) {
		t.Fatal("ack history did not evict oldest and retain newest IDs")
	}
}

func TestOutboundOptionsDefaults(t *testing.T) {
	opts := Options{}
	opts.setDefaults()
	if opts.OutboundQueueSize != 128 || opts.OutboundControlQueueSize != 32 {
		t.Fatalf("outbound queue defaults = %d/%d, want 128/32", opts.OutboundQueueSize, opts.OutboundControlQueueSize)
	}
	if opts.OutboundTrackedGlobalMaxBytes != 512<<20 {
		t.Fatalf("outbound tracked default = %d, want %d", opts.OutboundTrackedGlobalMaxBytes, 512<<20)
	}
	if opts.OutboundCriticalGlobalMaxBytes != 64<<20 {
		t.Fatalf("outbound critical default = %d, want %d", opts.OutboundCriticalGlobalMaxBytes, 64<<20)
	}
}

func TestServerNewConnectionsShareOutboundBudgetAndQueueLimits(t *testing.T) {
	srv := New(Options{
		OutboundQueueSize:              7,
		OutboundControlQueueSize:       3,
		OutboundTrackedGlobalMaxBytes:  20,
		OutboundCriticalGlobalMaxBytes: 30,
	})
	var rawKey crypto.Key
	key := rawKey.WithID()
	c1 := srv.newConn(nil, key, 1, 1)
	c2 := srv.newConn(nil, key, 2, 1)
	defer c1.Close()
	defer c2.Close()

	c1Ordinary := cap(c1.outbound) + cap(c1.outboundCritical) + cap(c1.outboundBulk)
	c2Ordinary := cap(c2.outbound) + cap(c2.outboundCritical) + cap(c2.outboundBulk)
	if c1Ordinary != 7 || cap(c1.outboundControl) != 3 || c2Ordinary != 7 || cap(c2.outboundControl) != 3 {
		t.Fatalf("server queue caps = %d/%d and %d/%d, want 7/3",
			c1Ordinary, cap(c1.outboundControl), c2Ordinary, cap(c2.outboundControl))
	}
	if c1.outboundTrackedBudget != srv.outboundTrackedBudget || c2.outboundTrackedBudget != srv.outboundTrackedBudget {
		t.Fatal("server connections did not receive the shared outbound tracking budget")
	}
	if got := srv.outboundTrackedBudget.maxBytes; got != 20 {
		t.Fatalf("server outbound tracked max = %d, want 20", got)
	}
	if c1.outboundCriticalTrackedBudget != srv.outboundCriticalBudget || c2.outboundCriticalTrackedBudget != srv.outboundCriticalBudget {
		t.Fatal("server connections did not receive the shared critical tracking budget")
	}
	if got := srv.outboundCriticalBudget.maxBytes; got != 30 {
		t.Fatalf("server outbound critical max = %d, want 30", got)
	}
}

func TestEncryptOutboundFrameDecryptsWithGotdCipher(t *testing.T) {
	var key crypto.Key
	if _, err := rand.Read(key[:]); err != nil {
		t.Fatalf("rand key: %v", err)
	}
	authKey := key.WithID()
	body := mustEncodeTL(t, &mt.NewSessionCreated{
		FirstMsgID: 111,
		UniqueID:   222,
		ServerSalt: 333,
	})
	c := &Conn{
		cipher:    crypto.NewServerCipher(rand.Reader),
		key:       authKey,
		salt:      12345,
		sessionID: 67890,
	}
	out, err := c.encryptOutboundFrame(&outboundFrame{
		msgID:  7649066000000000001,
		seqNo:  1,
		typeID: mt.NewSessionCreatedTypeID,
		body:   body,
	})
	if err != nil {
		t.Fatalf("encrypt: %v", err)
	}
	data, err := crypto.NewClientCipher(rand.Reader).DecryptFromBuffer(authKey, &bin.Buffer{Buf: append([]byte(nil), out.Raw()...)})
	if err != nil {
		t.Fatalf("decrypt: %v", err)
	}
	if data.Salt != c.salt || data.SessionID != c.sessionID {
		t.Fatalf("salt/session = %d/%d, want %d/%d", data.Salt, data.SessionID, c.salt, c.sessionID)
	}
	if got := data.Data(); !bytes.Equal(got, body) {
		t.Fatalf("body = %x, want %x", got, body)
	}
}

func TestOutboundActorSerializesConcurrentSends(t *testing.T) {
	const dc = 2
	addr, pub, srv := startTestServer(t, Options{DC: dc})
	conn, auth, cipher := dialHandshake(t, addr, dc, pub)

	clientMsgID := proto.NewMessageIDGen(time.Now)
	sendEncryptedWithSeq(t, conn, cipher, auth, clientMsgID.New(proto.MessageFromClient), 1, &mt.PingRequest{PingID: 1})
	collectReplies(t, conn, cipher, auth.AuthKey, mt.MsgsAckTypeID)
	freezeActiveTestSessionProfile(t, srv.Conns(), auth.AuthKey.ID, auth.SessionID, tlprofile.ProfileCanonical)
	srv.Conns().SetReceivesUpdates(auth.SessionID, true)

	const sends = 64
	var wg sync.WaitGroup
	errs := make(chan error, sends)
	for i := 0; i < sends; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			errs <- srv.Conns().PushToSession(ctx, auth.SessionID, proto.MessageFromServer, &tg.UpdatesTooLong{})
		}()
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("push: %v", err)
		}
	}

	var prevMsgID int64
	var prevSeqNo int32 = -1
	for i := 0; i < sends; i++ {
		data, id, _ := readServerMessage(t, conn, cipher, auth.AuthKey)
		if id != tg.UpdatesTooLongTypeID {
			t.Fatalf("message %d type = %#x, want updatesTooLong", i, id)
		}
		if i > 0 && data.MessageID <= prevMsgID {
			t.Fatalf("message %d msg_id = %d after %d, want strictly increasing", i, data.MessageID, prevMsgID)
		}
		if data.SeqNo%2 != 1 {
			t.Fatalf("message %d seq_no = %d, want odd content-related seq_no", i, data.SeqNo)
		}
		if i > 0 && data.SeqNo <= prevSeqNo {
			t.Fatalf("message %d seq_no = %d after %d, want increasing", i, data.SeqNo, prevSeqNo)
		}
		prevMsgID = data.MessageID
		prevSeqNo = data.SeqNo
	}
}

func TestOutboundWriteErrorTerminallyClosesWithoutActorDeadlock(t *testing.T) {
	tr := &failAfterTransport{}
	tr.failAt.Store(1)
	c := newOutboundFailureTestConn(t, tr)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := c.SendEncoded(ctx, proto.MessageFromServer, exactTestUpdatesTooLong(t, c)); err == nil {
		t.Fatal("Send unexpectedly succeeded")
	}
	select {
	case <-c.outboundDone:
	case <-time.After(time.Second):
		t.Fatal("outbound actor deadlocked while terminalizing its own write error")
	}
	if got := tr.closes.Load(); got != 1 {
		t.Fatalf("transport closes = %d, want 1", got)
	}
	if err := c.SendEncoded(ctx, proto.MessageFromServer, exactTestUpdatesTooLong(t, c)); !errors.Is(err, ErrConnClosed) {
		t.Fatalf("second Send err = %v, want ErrConnClosed", err)
	}
	if got := tr.sends.Load(); got != 1 {
		t.Fatalf("physical sends after terminal error = %d, want 1", got)
	}
}

func TestOutboundResendWriteErrorTerminallyCloses(t *testing.T) {
	tr := &failAfterTransport{}
	c := newOutboundFailureTestConn(t, tr)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	if err := c.SendEncoded(ctx, proto.MessageFromServer, exactTestUpdatesTooLong(t, c)); err != nil {
		t.Fatalf("initial Send: %v", err)
	}
	data, err := crypto.NewClientCipher(rand.Reader).DecryptFromBuffer(c.key, &bin.Buffer{Buf: tr.lastFrame()})
	if err != nil {
		t.Fatalf("decrypt initial frame: %v", err)
	}
	tr.failAt.Store(2)
	if _, err := c.ResendMessages(ctx, []int64{data.MessageID}); err == nil {
		t.Fatal("ResendMessages unexpectedly succeeded")
	}
	select {
	case <-c.outboundDone:
	case <-time.After(time.Second):
		t.Fatal("outbound actor did not exit after resend write error")
	}
	if got := tr.closes.Load(); got != 1 {
		t.Fatalf("transport closes = %d, want 1", got)
	}
}

func TestOutboundTrackedBudgetSharedAcrossConnections(t *testing.T) {
	budget := newOutboundTrackedBudget(12)
	tr1 := &failAfterTransport{}
	tr2 := &failAfterTransport{}
	c1 := newOutboundTestConn(t, tr1, budget)
	c2 := newOutboundTestConn(t, tr2, budget)
	body := exactTestUpdatesEncoded(t, c1, make([]byte, 8))
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	if err := c1.SendEncoded(ctx, proto.MessageFromServer, body); err != nil {
		t.Fatalf("first connection send: %v", err)
	}
	if got := budget.snapshot(); got != 8 {
		t.Fatalf("tracked bytes after first connection = %d, want 8", got)
	}
	if err := c2.SendEncoded(ctx, proto.MessageFromServer, body); !errors.Is(err, ErrOutboundTrackedBudget) && !errors.Is(err, ErrConnClosed) {
		t.Fatalf("second connection send err = %v, want tracked budget/closed", err)
	}
	select {
	case <-c2.outboundDone:
	case <-time.After(time.Second):
		t.Fatal("budget-exhausted connection did not terminate")
	}
	if got := tr2.sends.Load(); got != 0 {
		t.Fatalf("budget-exhausted connection wrote %d frames, want 0", got)
	}
	if got := budget.snapshot(); got != 8 {
		t.Fatalf("tracked bytes after second rejection = %d, want first connection's 8", got)
	}

	c1.Close()
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("tracked bytes after first connection close = %d, want 0", got)
	}
}

func TestOutboundTrackedBudgetReleaseBroadcastsToAllWaiters(t *testing.T) {
	const waiters = 8
	budget := newOutboundTrackedBudget(waiters)
	if !budget.reserve(waiters) {
		t.Fatal("reserve initial saturated budget")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	results := make(chan error, waiters)
	for i := 0; i < waiters; i++ {
		go func() {
			results <- budget.waitReserve(ctx, nil, 1)
		}()
	}

	deadline := time.Now().Add(time.Second)
	for {
		budget.wakeMu.Lock()
		got := budget.wake.waiters
		budget.wakeMu.Unlock()
		if got == waiters {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("subscribed waiters = %d, want %d", got, waiters)
		}
		time.Sleep(time.Millisecond)
	}

	// One batch release creates capacity for every waiter. A single-token notification strands
	// seven of them forever because successful reservations do not produce another wake-up.
	budget.release(waiters)
	for i := 0; i < waiters; i++ {
		if err := <-results; err != nil {
			t.Fatalf("waiter %d: %v", i, err)
		}
	}
	if got := budget.snapshot(); got != waiters {
		t.Fatalf("reserved bytes after broadcast = %d, want %d", got, waiters)
	}
	budget.release(waiters)
}

func TestOutboundGlobalBudgetIncludesQueuedBodies(t *testing.T) {
	budget := newOutboundTrackedBudget(24)
	tr := newBlockingOutboundTransport()
	c := newOutboundTestConn(t, tr, budget)
	body := exactTestUpdatesEncoded(t, c, make([]byte, 8))

	if err := c.SendBestEffortEncoded(context.Background(), proto.MessageFromServer, body, 0); err != nil {
		t.Fatalf("enqueue writing body: %v", err)
	}
	select {
	case <-tr.started:
	case <-time.After(time.Second):
		t.Fatal("outbound actor did not start blocked write")
	}
	for i := 0; i < 2; i++ {
		if err := c.SendBestEffortEncoded(context.Background(), proto.MessageFromServer, body, 0); err != nil {
			t.Fatalf("enqueue queued body %d: %v", i, err)
		}
	}
	if got := budget.snapshot(); got != 24 {
		t.Fatalf("writing + queued budget = %d, want 24", got)
	}
	if err := c.SendBestEffortEncoded(context.Background(), proto.MessageFromServer, body, 0); !errors.Is(err, ErrOutboundTrackedBudget) {
		t.Fatalf("over-budget enqueue err = %v, want ErrOutboundTrackedBudget", err)
	}
	select {
	case <-c.outboundDone:
		t.Fatal("best-effort global pressure terminated a healthy connection")
	case <-time.After(50 * time.Millisecond):
	}
	if got := budget.snapshot(); got != 24 {
		t.Fatalf("budget after non-terminal rejection = %d, want existing 24", got)
	}
	if err := tr.Close(); err != nil {
		t.Fatalf("close blocking transport: %v", err)
	}
	select {
	case <-c.outboundDone:
	case <-time.After(time.Second):
		t.Fatal("outbound actor did not stop after transport failure")
	}
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("budget after transport close = %d, want zero", got)
	}
}

func TestOutboundOversizedBodyRejectedBeforeEncryption(t *testing.T) {
	budget := newOutboundTrackedBudget(64 << 20)
	tr := &failAfterTransport{}
	c := newOutboundTestConn(t, tr, budget)
	body := exactTestUpdatesEncoded(t, c, make([]byte, maxOutboundBodyBytes+1))
	err := c.SendEncoded(context.Background(), proto.MessageFromServer, body)
	if !errors.Is(err, ErrOutboundMessageTooLarge) {
		t.Fatalf("oversized outbound err = %v, want ErrOutboundMessageTooLarge", err)
	}
	if got := tr.sends.Load(); got != 0 {
		t.Fatalf("oversized outbound wrote %d frames, want zero", got)
	}
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("oversized outbound reserved %d bytes, want zero", got)
	}
}

func TestOutboundCloseRaceDrainsEveryProducerReservation(t *testing.T) {
	budget := newOutboundTrackedBudget(1 << 20)
	c := newOutboundTestConn(t, &failAfterTransport{}, budget)
	body := exactTestUpdatesEncoded(t, c, make([]byte, 128))
	start := make(chan struct{})
	var wg sync.WaitGroup
	for i := 0; i < 128; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_ = c.SendBestEffortEncoded(context.Background(), proto.MessageFromServer, body, 0)
		}()
	}
	close(start)
	c.Close()
	wg.Wait()
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("outbound budget after close/enqueue race = %d, want zero", got)
	}
}

func TestOutboundTrackedBudgetAckAndCloseReturnExactly(t *testing.T) {
	t.Run("ack", func(t *testing.T) {
		budget := newOutboundTrackedBudget(64)
		tr := &failAfterTransport{}
		c := newOutboundTestConn(t, tr, budget)
		metrics := &acknowledgementCaptureMetrics{}
		c.metrics = metrics
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		body := exactTestUpdatesEncoded(t, c, make([]byte, 12))
		if err := c.SendEncoded(ctx, proto.MessageFromServer, body); err != nil {
			t.Fatalf("send: %v", err)
		}
		if got := budget.snapshot(); got != 12 {
			t.Fatalf("tracked bytes after send = %d, want 12", got)
		}
		data, err := crypto.NewClientCipher(rand.Reader).DecryptFromBuffer(c.key, &bin.Buffer{Buf: tr.lastFrame()})
		if err != nil {
			t.Fatalf("decrypt frame: %v", err)
		}
		// Windows wall-clock resolution can otherwise make an immediate ACK look
		// like zero retention even though sentAt was populated after the write.
		time.Sleep(time.Millisecond)
		c.AckServerMessages([]int64{data.MessageID})
		deadline := time.Now().Add(time.Second)
		for budget.snapshot() != 0 && time.Now().Before(deadline) {
			time.Sleep(time.Millisecond)
		}
		if got := budget.snapshot(); got != 0 {
			t.Fatalf("tracked bytes after ack = %d, want 0", got)
		}
		if got := metrics.count.Load(); got != 1 {
			t.Fatalf("logical ACK metric count = %d, want 1", got)
		}
		if got := metrics.bytes.Load(); got != 12 {
			t.Fatalf("logical ACK metric bytes = %d, want 12", got)
		}
		if metrics.retainedNS.Load() <= 0 {
			t.Fatal("logical ACK metric did not record positive retention")
		}
		if metrics.rpcResult.Load() {
			t.Fatal("ordinary update ACK was classified as rpc_result")
		}
	})

	t.Run("close", func(t *testing.T) {
		budget := newOutboundTrackedBudget(64)
		c := newOutboundTestConn(t, &failAfterTransport{}, budget)
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		body := exactTestUpdatesEncoded(t, c, make([]byte, 12))
		if err := c.SendEncoded(ctx, proto.MessageFromServer, body); err != nil {
			t.Fatalf("send: %v", err)
		}
		if got := budget.snapshot(); got != 12 {
			t.Fatalf("tracked bytes after send = %d, want 12", got)
		}
		c.Close()
		if got := budget.snapshot(); got != 0 {
			t.Fatalf("tracked bytes after close = %d, want 0", got)
		}
	})
}

func TestOutboundTrackedBudgetWriteFailureReturnsReservation(t *testing.T) {
	budget := newOutboundTrackedBudget(64)
	tr := &failAfterTransport{}
	tr.failAt.Store(1)
	c := newOutboundTestConn(t, tr, budget)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	body := exactTestUpdatesEncoded(t, c, make([]byte, 12))
	if err := c.SendEncoded(ctx, proto.MessageFromServer, body); err == nil {
		t.Fatal("send unexpectedly succeeded")
	}
	select {
	case <-c.outboundDone:
	case <-time.After(time.Second):
		t.Fatal("write-failed connection did not terminate")
	}
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("tracked bytes after write failure = %d, want 0", got)
	}
}

func TestOutboundStateCompactsImmutableRPCResultAndReplaysExactBody(t *testing.T) {
	budget := newOutboundTrackedBudget(1 << 20)
	state := newOutboundStateWithLimits(budget, 64, 1<<20)
	inner := bytes.Repeat([]byte{0x5a}, 4096)
	var body bin.Buffer
	body.PutID(proto.ResultTypeID)
	body.PutLong(7001)
	body.Put(inner)
	wire := body.Raw()
	if !budget.reserve(len(wire)) {
		t.Fatal("reserve first-write body")
	}
	frame := &outboundFrame{
		msgID:             9001,
		seqNo:             1,
		typeID:            proto.ResultTypeID,
		body:              wire,
		reservedBytes:     len(wire),
		reservationBudget: budget,
		reqMsgID:          7001,
		replaySource:      &staticRPCReplaySource{inner: append([]byte(nil), inner...)},
		innerDigest:       sha256.Sum256(inner),
		uncompressedBytes: len(inner),
		logicalBytes:      len(wire),
	}
	if err := state.admitReserved(frame); err != nil {
		t.Fatalf("admit frame: %v", err)
	}
	if !state.compactImmutableFrame(frame) {
		t.Fatal("immutable frame was not compacted")
	}
	if frame.body != nil {
		t.Fatal("compacted frame retained full body")
	}
	if got := budget.snapshot(); got != outboundReplayDescriptorCharge {
		t.Fatalf("retained bytes = %d, want descriptor charge %d", got, outboundReplayDescriptorCharge)
	}
	replay, ok := state.rpcResult(7001)
	if !ok || replay.replaySource == nil || len(replay.body) != 0 {
		t.Fatalf("replay descriptor = %+v ok=%v", replay, ok)
	}
	materialized, err := replay.materializeRPCResultBody(context.Background(), 7001)
	if err != nil {
		t.Fatalf("materialize replay: %v", err)
	}
	if !bytes.Equal(materialized, wire) {
		t.Fatal("materialized replay differs from first-write body")
	}
	state.ack([]int64{9001})
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("retained bytes after ACK = %d, want 0", got)
	}
}

func TestImmutableRPCResultMaterializesDirectlyIntoScratch(t *testing.T) {
	inner := bytes.Repeat([]byte{0x6b}, 1<<20)
	logicalBytes := 12 + len(inner)
	replay := &encodedOutboundMessage{
		typeID:            proto.ResultTypeID,
		reqMsgID:          7101,
		replaySource:      &staticRPCReplaySource{inner: inner},
		innerDigest:       sha256.Sum256(inner),
		uncompressedBytes: len(inner),
		logicalBytes:      logicalBytes,
	}
	pool := newOutboundReplayBodyPool([]outboundReplayBodyClassSpec{{size: logicalBytes, maxIdle: 1}})
	scratch, class := pool.acquire(logicalBytes)
	body, usedScratch, err := replay.materializeRPCResultBodyInto(context.Background(), replay.reqMsgID, scratch)
	if err != nil {
		t.Fatalf("materialize replay: %v", err)
	}
	if !usedScratch {
		t.Fatal("descriptor replay did not use the supplied scratch buffer")
	}
	if len(body) != logicalBytes || &body[0] != &scratch[:cap(scratch)][0] {
		t.Fatal("materialized body does not alias the supplied scratch buffer")
	}
	if got := int64(binary.LittleEndian.Uint64(body[4:12])); got != replay.reqMsgID {
		t.Fatalf("materialized req_msg_id = %d, want %d", got, replay.reqMsgID)
	}
	pool.release(class, body)
	if got := len(pool.classes[class].idle); got != 1 {
		t.Fatalf("idle pooled bodies = %d, want 1", got)
	}
}

func TestOutboundReplayBodyPoolBoundsIdleBuffers(t *testing.T) {
	pool := newOutboundReplayBodyPool([]outboundReplayBodyClassSpec{{size: 4096, maxIdle: 1}})
	first, firstClass := pool.acquire(4000)
	second, secondClass := pool.acquire(4000)
	if firstClass != 0 || secondClass != 0 || cap(first) != 4096 || cap(second) != 4096 {
		t.Fatalf("acquired classes/capacities = (%d,%d) (%d,%d)", firstClass, cap(first), secondClass, cap(second))
	}
	pool.release(firstClass, first)
	pool.release(secondClass, second)
	if got := len(pool.classes[0].idle); got != 1 {
		t.Fatalf("idle pooled bodies = %d, want bounded at 1", got)
	}
	oversized, class := pool.acquire(4097)
	if oversized != nil || class != -1 {
		t.Fatalf("oversized acquisition = len:%d class:%d, want GC-owned nil/-1", len(oversized), class)
	}
}

func BenchmarkImmutableRPCResultMaterializePooled(b *testing.B) {
	inner := bytes.Repeat([]byte{0x6b}, 1<<20)
	logicalBytes := 12 + len(inner)
	replay := &encodedOutboundMessage{
		typeID:            proto.ResultTypeID,
		reqMsgID:          7101,
		replaySource:      &staticRPCReplaySource{inner: inner},
		innerDigest:       sha256.Sum256(inner),
		uncompressedBytes: len(inner),
		logicalBytes:      logicalBytes,
	}
	pool := newOutboundReplayBodyPool([]outboundReplayBodyClassSpec{{size: logicalBytes, maxIdle: 1}})
	b.ReportAllocs()
	b.SetBytes(int64(logicalBytes))
	b.ResetTimer()
	for range b.N {
		scratch, class := pool.acquire(logicalBytes)
		body, usedScratch, err := replay.materializeRPCResultBodyInto(context.Background(), replay.reqMsgID, scratch)
		if err != nil || !usedScratch {
			b.Fatalf("materialize replay: used=%v err=%v", usedScratch, err)
		}
		pool.release(class, body)
	}
}

func TestOutboundBulkACKWindowWakesNextWaiter(t *testing.T) {
	state := newOutboundStateWithLimits(newOutboundTrackedBudget(1<<20), 128, 1<<20)
	leasing := make([]*outboundBulkCreditLease, 0, defaultBulkACKWindow+1)
	for range defaultBulkACKWindow + 1 {
		leasing = append(leasing, state.reserveBulkCredit())
	}
	woken := make(chan bool, 1)
	leasing[len(leasing)-1].credit.subscribe(func(success bool) { woken <- success })
	select {
	case <-woken:
		t.Fatal("window overflow waiter woke before ACK credit release")
	default:
	}
	leasing[0].releaseIfOwned()
	select {
	case success := <-woken:
		if !success {
			t.Fatal("window waiter was canceled instead of granted")
		}
	case <-time.After(time.Second):
		t.Fatal("window waiter did not wake after credit release")
	}
	for _, lease := range leasing[1:] {
		lease.releaseIfOwned()
	}
}

func TestOutboundStateEvictionReturnsTrackedBudget(t *testing.T) {
	budget := newOutboundTrackedBudget(64)
	state := newOutboundStateWithLimits(budget, 2, 8)
	defer state.releaseAll()
	frames := make([]*outboundFrame, 0, 3)
	for id := int64(1); id <= 3; id++ {
		frame := &outboundFrame{msgID: id, body: make([]byte, 4), reservedBytes: 4}
		frames = append(frames, frame)
		if !budget.reserve(len(frame.body)) {
			t.Fatalf("reserve frame %d", id)
		}
		dropped := state.addReserved(frame)
		if id < 3 && dropped != 0 {
			t.Fatalf("frame %d dropped %d, want 0", id, dropped)
		}
		if id == 3 && dropped != 1 {
			t.Fatalf("third frame dropped %d, want 1", dropped)
		}
	}
	if got := budget.snapshot(); got != 8 {
		t.Fatalf("tracked bytes after eviction = %d, want 8", got)
	}
	if frames[0].body != nil {
		t.Fatal("evicted frame retained its body reference")
	}
	state.releaseAll()
	if got := budget.snapshot(); got != 0 {
		t.Fatalf("tracked bytes after state close = %d, want 0", got)
	}
}

func TestOutboundStateReleasesMixedBodyAndControlBudgets(t *testing.T) {
	bodyBudget := newOutboundTrackedBudget(16)
	controlBudget := newOutboundTrackedBudget(16)
	state := newOutboundStateWithLimits(bodyBudget, 1, 16)

	if !controlBudget.reserve(4) {
		t.Fatal("reserve control frame")
	}
	controlFrame := &outboundFrame{
		msgID:             1,
		body:              make([]byte, 4),
		reservedBytes:     4,
		reservationBudget: controlBudget,
	}
	if dropped := state.addReserved(controlFrame); dropped != 0 {
		t.Fatalf("first add dropped %d, want 0", dropped)
	}

	if !bodyBudget.reserve(4) {
		t.Fatal("reserve body frame")
	}
	bodyFrame := &outboundFrame{
		msgID:             2,
		body:              make([]byte, 4),
		reservedBytes:     4,
		reservationBudget: bodyBudget,
	}
	if dropped := state.addReserved(bodyFrame); dropped != 1 {
		t.Fatalf("second add dropped %d, want control frame eviction", dropped)
	}
	if got := controlBudget.snapshot(); got != 0 {
		t.Fatalf("control budget after eviction = %d, want 0", got)
	}
	if got := bodyBudget.snapshot(); got != 4 {
		t.Fatalf("body budget after eviction = %d, want 4", got)
	}
	if controlFrame.body != nil || controlFrame.reservationBudget != nil {
		t.Fatal("evicted control frame retained body or budget ownership")
	}

	state.releaseAll()
	if got := bodyBudget.snapshot(); got != 0 {
		t.Fatalf("body budget after state close = %d, want 0", got)
	}
	if bodyFrame.body != nil || bodyFrame.reservationBudget != nil {
		t.Fatal("closed body frame retained body or budget ownership")
	}
}

func TestSendBestEffortQueueFullBehavior(t *testing.T) {
	c := &Conn{metrics: NopMetrics{}, outboundTrackedBudget: newOutboundTrackedBudget(1 << 20)}
	c.outbound = make(chan *outboundOp, 1)
	c.outboundControl = make(chan *outboundOp, 1)
	c.outboundStop = make(chan struct{})
	// 占满普通队列，模拟出站拥塞。
	c.outbound <- &outboundOp{}

	if err := c.SendBestEffort(context.Background(), proto.MessageFromServer, &mt.MsgsAck{}, 0); err != ErrOutboundQueueFull {
		t.Fatalf("timeout=0 on full queue: err = %v, want ErrOutboundQueueFull", err)
	}

	start := time.Now()
	if err := c.SendBestEffort(context.Background(), proto.MessageFromServer, &mt.MsgsAck{}, 30*time.Millisecond); err != ErrOutboundQueueFull {
		t.Fatalf("timeout=30ms on full queue: err = %v, want ErrOutboundQueueFull", err)
	}
	if waited := time.Since(start); waited < 30*time.Millisecond {
		t.Fatalf("timeout wait = %v, want >= 30ms", waited)
	}

	canceled, cancel := context.WithCancel(context.Background())
	cancel()
	if err := c.SendBestEffort(canceled, proto.MessageFromServer, &mt.MsgsAck{}, time.Second); err != context.Canceled {
		t.Fatalf("canceled ctx on full queue: err = %v, want context.Canceled", err)
	}

	// 腾出队列后快路径应直接入队成功。
	<-c.outbound
	if err := c.SendBestEffort(context.Background(), proto.MessageFromServer, &mt.MsgsAck{}, 0); err != nil {
		t.Fatalf("enqueue after drain: %v", err)
	}
	if got := len(c.outbound); got != 1 {
		t.Fatalf("queued ops = %d, want 1", got)
	}
}

func TestSendAsyncControlQueueBoundary(t *testing.T) {
	c := &Conn{metrics: NopMetrics{}, outboundTrackedBudget: newOutboundTrackedBudget(1 << 20)}
	c.outbound = make(chan *outboundOp, 1)
	c.outboundControl = make(chan *outboundOp, 1)
	c.outboundStop = make(chan struct{})
	c.outboundControl <- &outboundOp{kind: outboundAck}

	if err := c.SendAsync(context.Background(), proto.MessageFromServer, &mt.MsgsAck{}); err != nil {
		t.Fatalf("SendAsync on full control queue: %v", err)
	}
	if got := len(c.outboundControl); got != 1 {
		t.Fatalf("control queue len = %d, want bounded at 1", got)
	}
}

func TestFrameNeedsAckServiceExceptions(t *testing.T) {
	cases := []struct {
		name string
		id   uint32
		want bool
	}{
		{name: "pong", id: mt.PongTypeID, want: false},
		{name: "future_salts", id: mt.FutureSaltsTypeID, want: false},
		{name: "msgs_ack", id: mt.MsgsAckTypeID, want: false},
		{name: "updatesTooLong", id: tg.UpdatesTooLongTypeID, want: true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			if got := frameNeedsAck(tc.id); got != tc.want {
				t.Fatalf("frameNeedsAck(%s) = %v, want %v", tc.name, got, tc.want)
			}
		})
	}
}

func TestOutboundResendAndAckState(t *testing.T) {
	const dc = 2
	addr, pub, srv := startTestServer(t, Options{DC: dc})
	conn, auth, cipher := dialHandshake(t, addr, dc, pub)

	clientMsgID := proto.NewMessageIDGen(time.Now)
	sendEncryptedWithSeq(t, conn, cipher, auth, clientMsgID.New(proto.MessageFromClient), 1, &mt.PingRequest{PingID: 1})
	collectReplies(t, conn, cipher, auth.AuthKey, mt.MsgsAckTypeID)
	freezeActiveTestSessionProfile(t, srv.Conns(), auth.AuthKey.ID, auth.SessionID, tlprofile.ProfileCanonical)
	srv.Conns().SetReceivesUpdates(auth.SessionID, true)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	if err := srv.Conns().PushToSession(ctx, auth.SessionID, proto.MessageFromServer, &tg.UpdatesTooLong{}); err != nil {
		cancel()
		t.Fatalf("push: %v", err)
	}
	cancel()

	original, id, _ := readServerMessage(t, conn, cipher, auth.AuthKey)
	if id != tg.UpdatesTooLongTypeID {
		t.Fatalf("pushed type = %#x, want updatesTooLong", id)
	}

	resendReqID := clientMsgID.New(proto.MessageFromClient)
	sendEncryptedWithSeq(t, conn, cipher, auth, resendReqID, 3, &mt.MsgResendReq{MsgIDs: []int64{original.MessageID}})
	resent, resentType, _ := readServerMessage(t, conn, cipher, auth.AuthKey)
	if resentType != tg.UpdatesTooLongTypeID {
		t.Fatalf("resent type = %#x, want updatesTooLong", resentType)
	}
	if resent.MessageID != original.MessageID || resent.SeqNo != original.SeqNo {
		t.Fatalf("resent frame = (msg_id=%d seq=%d), want original (msg_id=%d seq=%d)",
			resent.MessageID, resent.SeqNo, original.MessageID, original.SeqNo)
	}
	_, stateType, stateBuf := readServerMessage(t, conn, cipher, auth.AuthKey)
	if stateType != mt.MsgsStateInfoTypeID {
		t.Fatalf("state type = %#x, want msgs_state_info", stateType)
	}
	assertStateInfo(t, stateBuf, resendReqID, []byte{msgStateReceived})
	_, ackType, _ := readServerMessage(t, conn, cipher, auth.AuthKey)
	if ackType != mt.MsgsAckTypeID {
		t.Fatalf("ack type = %#x, want msgs_ack", ackType)
	}

	sendEncryptedWithSeq(t, conn, cipher, auth, clientMsgID.New(proto.MessageFromClient), 4, &mt.MsgsAck{MsgIDs: []int64{original.MessageID}})
	ackedResendReqID := clientMsgID.New(proto.MessageFromClient)
	sendEncryptedWithSeq(t, conn, cipher, auth, ackedResendReqID, 5, &mt.MsgResendReq{MsgIDs: []int64{original.MessageID}})
	_, ackedStateType, ackedStateBuf := readServerMessage(t, conn, cipher, auth.AuthKey)
	if ackedStateType != mt.MsgsStateInfoTypeID {
		t.Fatalf("after ack type = %#x, want msgs_state_info without resend", ackedStateType)
	}
	assertStateInfo(t, ackedStateBuf, ackedResendReqID, []byte{msgStateReceived})
}

func assertStateInfo(t *testing.T, b *bin.Buffer, reqMsgID int64, want []byte) {
	t.Helper()
	var info mt.MsgsStateInfo
	if err := info.Decode(b); err != nil {
		t.Fatalf("decode msgs_state_info: %v", err)
	}
	if info.ReqMsgID != reqMsgID {
		t.Fatalf("msgs_state_info.req_msg_id = %d, want %d", info.ReqMsgID, reqMsgID)
	}
	if string(info.Info) != string(want) {
		t.Fatalf("msgs_state_info.info = %v, want %v", []byte(info.Info), want)
	}
}