internal/mtprotoedge/inbound_preflight.go
Rama main · réplica de lectura del repositorio original.
package mtprotoedge
import (
"context"
"encoding/binary"
"errors"
"fmt"
"io"
"time"
"go.uber.org/zap"
"github.com/iamxvbaba/td/bin"
"github.com/iamxvbaba/td/mt"
"github.com/iamxvbaba/td/proto"
"github.com/iamxvbaba/td/tlprofile"
)
type inboundItemKind uint8
const (
inboundItemDuplicate inboundItemKind = iota + 1
inboundItemServiceDuplicate
inboundItemPing
inboundItemPingDelay
inboundItemFutureSalts
inboundItemMsgsAck
inboundItemStateReq
inboundItemResendReq
inboundItemStateInfo
inboundItemAllInfo
inboundItemDestroySession
inboundItemHTTPWait
inboundItemDropAnswer
inboundItemDestroyAuthKey
// inboundItemHelpTest carries the legacy help.test#c0e202f7 = Bool method,
// which was removed from the published schema but is still sent by some
// clients. It is answered with a fixed boolTrue and never reaches routing.
inboundItemHelpTest
inboundItemRPC
inboundItemCapacityError
inboundItemRPCAdmissionError
// inboundItemRewrappedRPC is an initConnection retry whose exact inner TL
// request is already executing (or completed) under the client's old msg_id.
// It never dispatches business code a second time.
inboundItemRewrappedRPC
// inboundItemReplayRPC is a request first observed by this physical Conn whose
// terminal result already exists in the cross-connection execution ledger. It is distinct
// from inboundItemDuplicate: a duplicate already present in this Conn's seen
// table must only be ACKed. The original owner/result is already using the same
// reliable TCP stream, so replaying it once per retransmit wave amplifies a
// client retry burst and can starve newer request IDs behind orphan results.
inboundItemReplayRPC
)
type inboundItem struct {
kind inboundItemKind
msgID int64
admissionSeq uint64
seqNo int32
typeID uint32
content bool
body []byte
payload any
admitted tlprofile.Admission
method string
replayAfterSuccessfulDelivery func() error
layerProfileEvidenceFreshness inboundLayerProfileEvidenceFreshness
}
type inboundLayerProfileEvidenceFreshness uint8
const (
// Unspecified is retained for focused force-style unit tests which construct
// inboundItem directly, outside MTProto envelope preflight. Production items
// are always classified from the frame's one clock sample.
inboundLayerProfileEvidenceFreshnessUnspecified inboundLayerProfileEvidenceFreshness = iota
inboundLayerProfileEvidenceFresh
inboundLayerProfileEvidenceRequestBound
)
func (i inboundItem) profileEvidenceFresh() bool {
return i.layerProfileEvidenceFreshness != inboundLayerProfileEvidenceRequestBound
}
type stagedClientMessage struct {
msgID int64
seqNo int32
content bool
service bool
}
type inboundPlan struct {
items []inboundItem
staged []stagedClientMessage
ackIDs []int64
logicalMin int64
releases []func()
// gzipExpandedBytes is the non-refundable per-frame decompression work
// already performed by outer and exact-layer nested gzip envelopes. Memory
// reservations are released when their buffers die, but this cumulative
// counter prevents sibling RPCs from recycling the same CPU budget.
gzipExpandedBytes int
rpcReservation *inboundRPCBatchReservation
rpcTasks []inboundRPC
rpcOwners []*rpcResultOwnerLease
rewrapAliases []*rpcRewrapAlias
}
func (p *inboundPlan) close() {
if p == nil {
return
}
// Drop exact typed request graphs and uncommitted task closures before their
// materialization reservation becomes reusable. Otherwise an abort could
// advertise the same bytes to another connection while this plan still kept
// the old graph reachable until its caller returned.
for i := range p.items {
p.items[i].admitted = tlprofile.Admission{}
}
for i := range p.rpcTasks {
p.rpcTasks[i] = inboundRPC{}
}
p.rpcTasks = nil
if p.rpcReservation != nil {
p.rpcReservation.abort()
p.rpcReservation = nil
}
for _, owner := range p.rpcOwners {
owner.Abort()
}
p.rpcOwners = nil
for _, alias := range p.rewrapAliases {
alias.releaseCandidate()
if alias != nil && alias.newOwner != nil {
alias.newOwner.Abort()
}
}
p.rewrapAliases = nil
for i := len(p.releases) - 1; i >= 0; i-- {
p.releases[i]()
}
p.releases = nil
}
func (p *inboundPlan) commitRewrapAliases(s *Server) error {
if p == nil || len(p.rewrapAliases) == 0 {
return nil
}
for i, alias := range p.rewrapAliases {
if err := alias.activate(s); err != nil {
for _, pending := range p.rewrapAliases[i:] {
pending.releaseCandidate()
if pending != nil && pending.newOwner != nil {
pending.newOwner.Abort()
}
}
p.rewrapAliases = nil
return err
}
}
p.rewrapAliases = nil
return nil
}
// rejectNewRPCOwners turns only ownership acquired by this batch into bounded
// capacity responses. Existing completed replays and pending joins remain
// active: canceling them would either lose a response or publish into another
// request's flight.
func (p *inboundPlan) rejectNewRPCOwners(indices []int) {
if p == nil {
return
}
for _, index := range indices {
if index >= 0 && index < len(p.items) {
p.items[index].kind = inboundItemCapacityError
}
}
kept := p.rewrapAliases[:0]
for _, alias := range p.rewrapAliases {
if alias == nil || alias.newOwner == nil {
kept = append(kept, alias)
continue
}
if alias.itemIndex >= 0 && alias.itemIndex < len(p.items) {
p.items[alias.itemIndex].kind = inboundItemCapacityError
p.items[alias.itemIndex].payload = nil
}
alias.releaseCandidate()
// This owner was acquired by the rejected batch and is not present in
// plan.rpcOwners because it belonged to a rewrap alias. Abort it here
// before dropping the alias, otherwise the exact flight remains pending
// forever with no task or publisher able to complete it.
alias.newOwner.Abort()
alias.newOwner = nil
}
p.rewrapAliases = kept
}
func (p *inboundPlan) commitRPCBatch() error {
if p == nil || p.rpcReservation == nil {
return nil
}
// handleEncrypted calls this only after ownership/session-control barriers,
// connState commit and every synchronous service action have completed. The
// batch is runnable immediately; using the old deferred scheduler token here
// was not a real barrier on a busy Conn because an existing ready token could
// dequeue newly appended tasks before activateRPCBatch ran.
err := p.rpcReservation.commit(p.rpcTasks)
if err != nil {
return err
}
p.rpcReservation = nil
p.rpcTasks = nil
// Every owner is now attached to exactly one queued task. Its task.release
// path aborts if no terminal rpc_result was published.
p.rpcOwners = nil
return nil
}
func (p *inboundPlan) includeLogicalID(msgID int64) {
if p.logicalMin == 0 || msgID < p.logicalMin {
p.logicalMin = msgID
}
}
func (p *inboundPlan) commitState(cs *connState) {
for _, m := range p.staged {
cs.trackInbound(m.msgID, m.seqNo, m.content, m.service, msgStateReceived)
}
}
type connStateOverlay struct {
base *connState
staged []stagedClientMessage
maxContentMsgID int64
maxContentSeqNo int32
}
func newConnStateOverlay(base *connState) connStateOverlay {
return connStateOverlay{
base: base,
maxContentMsgID: base.maxContentMsgID,
maxContentSeqNo: base.maxContentSeqNo,
}
}
func (o *connStateOverlay) seenRecord(msgID int64) (clientMsgRecord, bool) {
for i := len(o.staged) - 1; i >= 0; i-- {
m := o.staged[i]
if m.msgID == msgID {
return clientMsgRecord{state: msgStateReceived, seqNo: m.seqNo, content: m.content, service: m.service}, true
}
}
return o.base.seenRecord(msgID)
}
func (o *connStateOverlay) validateSeq(msgID int64, seqNo int32, content bool) int {
if !content {
return 0
}
if msgID > o.maxContentMsgID && seqNo > o.maxContentSeqNo {
return 0
}
for seenMsgID, record := range o.base.seen {
if !record.content {
continue
}
if seenMsgID < msgID && record.seqNo >= seqNo {
return badMsgSeqTooLow
}
if seenMsgID > msgID && record.seqNo <= seqNo {
return badMsgSeqTooHigh
}
}
for _, record := range o.staged {
if !record.content {
continue
}
if record.msgID < msgID && record.seqNo >= seqNo {
return badMsgSeqTooLow
}
if record.msgID > msgID && record.seqNo <= seqNo {
return badMsgSeqTooHigh
}
}
return 0
}
func (o *connStateOverlay) stage(msgID int64, seqNo int32, content, service bool) {
o.staged = append(o.staged, stagedClientMessage{msgID: msgID, seqNo: seqNo, content: content, service: service})
if content {
if msgID > o.maxContentMsgID {
o.maxContentMsgID = msgID
}
if seqNo > o.maxContentSeqNo {
o.maxContentSeqNo = seqNo
}
}
}
type inboundScope struct {
insideContainer bool
mustBeDuplicate bool
}
type inboundPreflightBudget struct {
depth int
containerDepth int
expanded int
now time.Time
}
func (s *Server) preflightInbound(cs *connState, msgID int64, seqNo int32, body []byte) (*inboundPlan, error) {
plan := &inboundPlan{logicalMin: 0}
overlay := newConnStateOverlay(cs)
// All envelope checks in one transport frame use the same clock sample. Apart
// from making boundary behavior deterministic, this lets walkInbound reject
// an invalid outer msg_id before spending CPU or memory on gzip expansion.
budget := &inboundPreflightBudget{now: s.clock.Now()}
if err := s.walkInbound(plan, &overlay, msgID, seqNo, body, inboundScope{}, budget); err != nil {
plan.close()
return nil, err
}
plan.gzipExpandedBytes = budget.expanded
plan.staged = overlay.staged
if plan.logicalMin == 0 {
plan.close()
return nil, fmt.Errorf("inbound plan accepted no logical message")
}
for _, item := range plan.items {
if item.kind == inboundItemDestroyAuthKey && len(plan.items) != 1 {
plan.close()
return nil, &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
}
return plan, nil
}
func (s *Server) walkInbound(
plan *inboundPlan,
overlay *connStateOverlay,
msgID int64,
seqNo int32,
body []byte,
scope inboundScope,
budget *inboundPreflightBudget,
) error {
if budget.depth > maxDispatchDepth {
return fmt.Errorf("mtproto wrapper depth %d exceeds %d", budget.depth, maxDispatchDepth)
}
b := &bin.Buffer{Buf: body}
typeID, err := b.PeekID()
if err != nil {
return fmt.Errorf("peek type id: %w", err)
}
if code := validateInboundMessageID(budget.now, msgID, scope.insideContainer); code != 0 {
if scope.insideContainer {
code = badMsgContainer
}
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
}
// A repeated outer container still has to be decoded far enough to enumerate
// its inner request ids, but none of those already-accepted inner bodies needs
// decoding again. In particular, never reinflate a gzip body merely to discover
// the content bit we already retained with its msg_id. If an outer duplicate
// introduces an unseen inner id, reject the container before touching its body.
// A seen top-level content gzip is also safe to short-circuit: a container is
// non-content, so it cannot be hidden behind that retained record. Ambiguous
// non-content top-level gzip still expands to distinguish a container wrapper.
if record, seen := overlay.seenRecord(msgID); scope.mustBeDuplicate ||
(typeID == proto.GZIPTypeID && seen && (scope.insideContainer || record.content)) {
if !seen || record.seqNo != seqNo {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
return appendInboundDuplicate(plan, msgID, seqNo, typeID, record)
}
if typeID == proto.GZIPTypeID {
data, release, err := s.decodeGZIPWithGlobalBudget(b)
if err != nil {
return fmt.Errorf("decode gzip: %w", err)
}
plan.releases = append(plan.releases, release)
budget.expanded += len(data)
if budget.expanded > maxDispatchExpandedBytes {
return fmt.Errorf("cumulative gzip expansion %d exceeds %d", budget.expanded, maxDispatchExpandedBytes)
}
budget.depth++
err = s.walkInbound(plan, overlay, msgID, seqNo, data, scope, budget)
budget.depth--
return err
}
if typeID == proto.MessageContainerTypeID {
if scope.insideContainer || budget.containerDepth != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
if code := validateClientEnvelope(budget.now, msgID, seqNo, typeID); code != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
}
outerRecord, outerSeen := overlay.seenRecord(msgID)
if outerSeen {
if outerRecord.content || outerRecord.seqNo != seqNo {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
} else {
if code := overlay.validateSeq(msgID, seqNo, false); code != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
}
overlay.stage(msgID, seqNo, false, false)
}
count, err := containerMessageCount(b)
if err != nil {
return fmt.Errorf("decode container count: %w", err)
}
if count > maxContainerMessages {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
container, release, err := s.decodeMessageContainerViews(b, count)
if err != nil {
return fmt.Errorf("decode container: %w", err)
}
plan.releases = append(plan.releases, release)
plan.items = growInboundItems(plan.items, count)
plan.ackIDs = growInt64s(plan.ackIDs, count)
overlay.staged = growStagedMessages(overlay.staged, count+1)
if len(container.Messages) == 0 {
plan.includeLogicalID(msgID)
return nil
}
budget.depth++
budget.containerDepth++
for _, m := range container.Messages {
if m.ID >= msgID || int32(m.SeqNo) > seqNo {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
if err := s.walkInbound(plan, overlay, m.ID, int32(m.SeqNo), m.Body, inboundScope{
insideContainer: true,
mustBeDuplicate: outerSeen,
}, budget); err != nil {
return err
}
}
budget.containerDepth--
budget.depth--
return nil
}
if scope.insideContainer {
if code := validateClientContainerEnvelope(msgID, seqNo, typeID); code != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
} else if code := validateClientEnvelope(budget.now, msgID, seqNo, typeID); code != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
}
content := clientMessageIsContentRelated(typeID, seqNo)
if record, seen := overlay.seenRecord(msgID); seen {
if record.seqNo != seqNo || record.content != content {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
return appendInboundDuplicate(plan, msgID, seqNo, typeID, record)
}
if scope.mustBeDuplicate {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
}
if code := overlay.validateSeq(msgID, seqNo, content); code != 0 {
return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
}
overlay.stage(msgID, seqNo, content, inboundTypeIsService(typeID))
plan.includeLogicalID(msgID)
item, err := preflightInboundItem(msgID, seqNo, typeID, content, body)
if err != nil {
return err
}
if validateInboundMessageID(budget.now, msgID, false) == 0 {
item.layerProfileEvidenceFreshness = inboundLayerProfileEvidenceFresh
} else {
// Inner container messages deliberately bypass the wall-clock rejection
// above, but old/future ids are request-bound and cannot publish mutable
// Layer/init/readiness/auth-bind evidence.
item.layerProfileEvidenceFreshness = inboundLayerProfileEvidenceRequestBound
}
plan.items = append(plan.items, item)
if content {
plan.ackIDs = append(plan.ackIDs, msgID)
}
return nil
}
func validateInboundMessageID(now time.Time, msgID int64, insideContainer bool) int {
if !validClientMessageIDBits(msgID) {
return badMsgIDInvalidBits
}
// A container's outer envelope supplies the wall-clock admission boundary for
// its inner messages. Inner ids still need the client low bits checked before
// wrapper expansion, but intentionally keep the established no-time-check rule.
if insideContainer {
return 0
}
msgTime := proto.MessageID(msgID).Time()
if msgTime.Before(now.Add(-300 * time.Second)) {
return badMsgIDTooLow
}
if msgTime.After(now.Add(30 * time.Second)) {
return badMsgIDTooHigh
}
return 0
}
func validClientMessageIDBits(msgID int64) bool {
return msgID > 0 && uint32(msgID) != 0 &&
proto.MessageID(msgID).Type() == proto.MessageFromClient
}
func appendInboundDuplicate(plan *inboundPlan, msgID int64, seqNo int32, typeID uint32, record clientMsgRecord) error {
plan.includeLogicalID(msgID)
kind := inboundItemDuplicate
if record.service {
kind = inboundItemServiceDuplicate
}
plan.items = append(plan.items, inboundItem{
kind: kind, msgID: msgID, seqNo: seqNo, typeID: typeID, content: record.content,
})
if record.content {
plan.ackIDs = append(plan.ackIDs, msgID)
}
return nil
}
func inboundTypeIsService(typeID uint32) bool {
switch typeID {
case mt.PingRequestTypeID,
mt.PingDelayDisconnectRequestTypeID,
mt.GetFutureSaltsRequestTypeID,
mt.MsgsAckTypeID,
mt.MsgsStateReqTypeID,
mt.MsgResendReqTypeID,
mt.MsgsStateInfoTypeID,
mt.MsgsAllInfoTypeID,
mt.DestroySessionRequestTypeID,
mt.HTTPWaitRequestTypeID,
mt.RPCDropAnswerRequestTypeID,
destroyAuthKeyRequestTypeID,
helpTestRequestTypeID:
return true
default:
return false
}
}
func growInboundItems(items []inboundItem, extra int) []inboundItem {
if extra <= cap(items)-len(items) {
return items
}
grown := make([]inboundItem, len(items), len(items)+extra)
copy(grown, items)
return grown
}
func growStagedMessages(items []stagedClientMessage, extra int) []stagedClientMessage {
if extra <= cap(items)-len(items) {
return items
}
grown := make([]stagedClientMessage, len(items), len(items)+extra)
copy(grown, items)
return grown
}
func growInt64s(items []int64, extra int) []int64 {
if extra <= cap(items)-len(items) {
return items
}
grown := make([]int64, len(items), len(items)+extra)
copy(grown, items)
return grown
}
type stateInfoPayload struct {
reqMsgID int64
info []byte
}
type allInfoPayload struct {
count int
info []byte
}
// int64VectorView is a bounded, immutable view of a TL Vector<long>. Keeping the
// encoded bytes avoids one retained []int64 allocation per service message while
// a whole container is in preflight. Execution materializes the small bounded
// slice only for the existing Conn APIs that require []int64, after all wrappers
// and every sibling message have already passed structural validation.
type int64VectorView struct {
raw []byte
count int
}
func decodeInt64VectorView(b *bin.Buffer, expectedTypeID uint32, max int) (int64VectorView, error) {
if b == nil || len(b.Buf) < 12 {
return int64VectorView{}, io.ErrUnexpectedEOF
}
if got := binary.LittleEndian.Uint32(b.Buf[:4]); got != expectedTypeID {
return int64VectorView{}, fmt.Errorf("unexpected constructor %#x", got)
}
if got := binary.LittleEndian.Uint32(b.Buf[4:8]); got != bin.TypeVector {
return int64VectorView{}, fmt.Errorf("unexpected vector constructor %#x", got)
}
count := int(int32(binary.LittleEndian.Uint32(b.Buf[8:12])))
if count < 0 {
return int64VectorView{}, fmt.Errorf("negative vector count %d", count)
}
if count > max {
return int64VectorView{}, fmt.Errorf("vector count %d exceeds %d", count, max)
}
if count > (len(b.Buf)-12)/8 {
return int64VectorView{}, io.ErrUnexpectedEOF
}
end := 12 + count*8
if end != len(b.Buf) {
return int64VectorView{}, fmt.Errorf("long vector has %d trailing bytes", len(b.Buf)-end)
}
return int64VectorView{raw: b.Buf[12:end:end], count: count}, nil
}
func (v int64VectorView) materialize() []int64 {
if v.count == 0 {
return nil
}
ids := make([]int64, v.count)
for i := range ids {
offset := i * 8
ids[i] = int64(binary.LittleEndian.Uint64(v.raw[offset : offset+8]))
}
return ids
}
func decodeInboundServiceExact(b *bin.Buffer, decoder bin.Decoder) error {
if err := decoder.Decode(b); err != nil {
return err
}
if b.Len() != 0 {
return fmt.Errorf("service message has %d trailing bytes", b.Len())
}
return nil
}
func preflightInboundItem(msgID int64, seqNo int32, typeID uint32, content bool, body []byte) (inboundItem, error) {
item := inboundItem{msgID: msgID, seqNo: seqNo, typeID: typeID, content: content, body: body}
b := &bin.Buffer{Buf: body}
switch typeID {
case mt.PingRequestTypeID:
var value mt.PingRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode ping: %w", err)
}
item.kind, item.payload = inboundItemPing, value
case mt.PingDelayDisconnectRequestTypeID:
var value mt.PingDelayDisconnectRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode ping_delay_disconnect: %w", err)
}
item.kind, item.payload = inboundItemPingDelay, value
case mt.GetFutureSaltsRequestTypeID:
var value mt.GetFutureSaltsRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode get_future_salts: %w", err)
}
item.kind, item.payload = inboundItemFutureSalts, value
case mt.MsgsAckTypeID:
value, err := decodeInt64VectorView(b, mt.MsgsAckTypeID, maxServiceMessageIDs)
if err != nil {
return item, fmt.Errorf("decode msgs_ack: %w", err)
}
item.kind, item.payload = inboundItemMsgsAck, value
case mt.MsgsStateReqTypeID:
value, err := decodeInt64VectorView(b, mt.MsgsStateReqTypeID, maxServiceMessageIDs)
if err != nil {
return item, fmt.Errorf("decode msgs_state_req: %w", err)
}
item.kind, item.payload = inboundItemStateReq, value
case mt.MsgResendReqTypeID:
value, err := decodeInt64VectorView(b, mt.MsgResendReqTypeID, maxServiceMessageIDs)
if err != nil {
return item, fmt.Errorf("decode msg_resend_req: %w", err)
}
item.kind, item.payload = inboundItemResendReq, value
case mt.MsgsStateInfoTypeID:
reqMsgID, info, err := msgsStateInfoView(b)
if err != nil {
return item, fmt.Errorf("decode msgs_state_info: %w", err)
}
item.kind, item.payload = inboundItemStateInfo, stateInfoPayload{reqMsgID: reqMsgID, info: info}
case mt.MsgsAllInfoTypeID:
count, info, err := msgsAllInfoView(b)
if err != nil {
return item, fmt.Errorf("decode msgs_all_info: %w", err)
}
if len(info) != count {
return item, fmt.Errorf("decode msgs_all_info: info length %d does not match msg_ids %d", len(info), count)
}
item.kind, item.payload = inboundItemAllInfo, allInfoPayload{count: count, info: info}
case mt.DestroySessionRequestTypeID:
var value mt.DestroySessionRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode destroy_session: %w", err)
}
item.kind, item.payload = inboundItemDestroySession, value
case mt.HTTPWaitRequestTypeID:
var value mt.HTTPWaitRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode http_wait: %w", err)
}
item.kind, item.payload = inboundItemHTTPWait, value
case mt.RPCDropAnswerRequestTypeID:
var value mt.RPCDropAnswerRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, fmt.Errorf("decode rpc_drop_answer: %w", err)
}
item.kind, item.payload = inboundItemDropAnswer, value
case destroyAuthKeyRequestTypeID:
var value destroyAuthKeyRequest
if err := decodeInboundServiceExact(b, &value); err != nil {
return item, err
}
item.kind, item.payload = inboundItemDestroyAuthKey, value
case helpTestRequestTypeID:
if len(body) != bin.Word {
return item, fmt.Errorf("decode help.test: expected %d bytes, got %d", bin.Word, len(body))
}
item.kind, item.payload = inboundItemHelpTest, helpTestRequest{}
default:
item.kind = inboundItemRPC
}
return item, nil
}
// prepareInboundRPCBatch performs the whole container's count/byte admission
// before copying or scheduling any API RPC. Capacity exhaustion is converted
// into one consistent local 500 WORKER_BUSY_TOO_LONG_RETRY result per new
// RPC; no business handler from the batch is allowed to start in that case.
func (s *Server) prepareInboundRPCBatch(ctx context.Context, c *Conn, plan *inboundPlan) error {
if s.layerRPC != nil {
return s.prepareInboundLayerRPCBatch(ctx, c, plan)
}
// Keep service-only frames (ping/ack/http_wait) allocation-free here. These
// collections are needed only after the first real API RPC acquires ownership.
var indices []int
var specs []inboundRPCSpec
var ownersInPlan map[int64]*rpcResultOwnerLease
flightCapacity := false
clearedPostInitCandidates := false
for i := range plan.items {
item := &plan.items[i]
if item.kind != inboundItemRPC && item.kind != inboundItemDuplicate {
continue
}
localDuplicate := item.kind == inboundItemDuplicate
if localDuplicate {
// connState already proves this msg_id was admitted on this physical
// generation. Its original owner either still holds the flight or has
// published a result after physical write; do not consume a global flight
// slot merely to ACK the duplicate.
continue
}
method := s.typeName(item.typeID)
init, isInitRewrap := decodeRPCRewrapInit(item.body)
if isInitRewrap {
firstInit := false
if item.profileEvidenceFresh() {
firstInit = !c.rpcRewrapInitialized.Swap(true)
c.setLegacyClientLayer(init.layer)
}
if candidate := s.rpcRewrap.claim(c, init.inner); candidate != nil {
claim, err := s.rpcResults.Acquire(c.authKeyID, c.sessionID, item.msgID)
if errors.Is(err, ErrRPCResultFlightCapacity) {
s.rpcRewrap.release(candidate)
c.metrics.InboundRPCDropped(candidate.method, "flight_capacity")
flightCapacity = true
item.kind = inboundItemCapacityError
continue
}
if err != nil {
s.rpcRewrap.release(candidate)
return err
}
switch claim.state {
case rpcResultAcquireCompleted:
s.rpcRewrap.commit(candidate)
item.kind = inboundItemReplayRPC
item.payload = claim.encoded
case rpcResultAcquireAcknowledged:
// The client already ACKed the correlated rpc_result. Retire the
// stale rewrap candidate and keep this duplicate ACK-only; neither
// the old body nor the business handler may run again.
s.rpcRewrap.commit(candidate)
item.kind = inboundItemDuplicate
item.payload = nil
case rpcResultAcquirePending:
s.rpcRewrap.commit(candidate)
item.kind = inboundItemRewrappedRPC
item.payload = claim.waiter
plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
conn: c, itemIndex: i, newReqID: item.msgID, method: candidate.method,
oldWaiter: claim.waiter, observeInit: firstInit, init: init,
})
case rpcResultAcquireOwner:
if ownersInPlan == nil {
ownersInPlan = make(map[int64]*rpcResultOwnerLease)
}
ownersInPlan[item.msgID] = claim.owner
item.kind = inboundItemRewrappedRPC
item.payload = claim.owner
plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
conn: c, itemIndex: i, newReqID: item.msgID, method: candidate.method,
oldWaiter: candidate.waiter, newOwner: claim.owner,
sourceConn: candidate.source, sourceOwner: candidate.owner,
observeInit: firstInit, init: init,
candidate: candidate, registry: s.rpcRewrap,
})
default:
s.rpcRewrap.release(candidate)
return ErrRPCResultFlightInvalid
}
s.log.Info("RPC init rewrap matched",
zap.String("method", candidate.method),
zap.Int64("old_req_msg_id", candidate.reqMsgID),
zap.Int64("new_req_msg_id", item.msgID),
zap.Bool("same_connection", candidate.source == c),
zap.String("auth_key_id", c.authKeyHex), zap.Int64("session_id", c.sessionID))
continue
}
} else if item.profileEvidenceFresh() && c.rpcRewrapInitialized.Load() && !clearedPostInitCandidates {
// A naked request after this connection has observed initConnection is
// event-level proof that the client finished moving its old running set.
// Retire any unmatched candidates without a timer.
s.rpcRewrap.clearSession(c)
clearedPostInitCandidates = true
}
claim, err := s.rpcResults.Acquire(c.authKeyID, c.sessionID, item.msgID)
if errors.Is(err, ErrRPCResultFlightCapacity) {
if item.kind == inboundItemRPC {
c.metrics.InboundRPCDropped(method, "flight_capacity")
flightCapacity = true
item.kind = inboundItemCapacityError
}
continue
}
if err != nil {
return err
}
switch claim.state {
case rpcResultAcquireCompleted:
s.log.Info("RPC duplicate replay from logical-session outbox",
zap.String("method", method),
zap.Int64("msg_id", item.msgID),
zap.String("auth_key_id", c.authKeyHex),
zap.Int64("session_id", c.sessionID),
)
item.kind = inboundItemReplayRPC
item.payload = claim.encoded
case rpcResultAcquireAcknowledged:
item.kind = inboundItemDuplicate
item.payload = nil
case rpcResultAcquirePending:
// A malformed/replayed container may repeat the same msg_id after this
// very plan installed its owner. More generally, any request already in
// this Conn's seen table shares the owner's reliable response path. Only
// a fresh physical replacement may wait for and replay the old result.
if ownersInPlan[item.msgID] != nil {
item.kind = inboundItemDuplicate
item.payload = nil
} else {
item.kind = inboundItemRewrappedRPC
item.payload = claim.waiter
plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
conn: c, itemIndex: i, newReqID: item.msgID, method: method, oldWaiter: claim.waiter,
})
}
case rpcResultAcquireOwner:
if item.kind == inboundItemDuplicate {
// connState says this request was already accepted, so absence from
// both completed and in-flight tables is not authority to execute it.
claim.owner.Abort()
continue
}
if ownersInPlan == nil {
ownersInPlan = make(map[int64]*rpcResultOwnerLease)
indices = make([]int, 0, len(plan.items))
specs = make([]inboundRPCSpec, 0, len(plan.items))
}
ownersInPlan[item.msgID] = claim.owner
plan.rpcOwners = append(plan.rpcOwners, claim.owner)
item.payload = claim.owner
indices = append(indices, i)
specs = append(specs, inboundRPCSpec{method: method, size: len(item.body)})
if !c.rpcRewrapInitialized.Load() && !isInitRewrap {
s.rpcRewrap.register(c, item.body, item.msgID, method, claim.owner)
}
default:
return ErrRPCResultFlightInvalid
}
}
if flightCapacity {
// One container is one API admission unit. If the cross-connection
// exactly-once table cannot claim every fresh request, none of this
// batch may reach a business handler.
for _, index := range indices {
plan.items[index].kind = inboundItemCapacityError
}
plan.rejectNewRPCOwners(indices)
return nil
}
if len(specs) == 0 {
return nil
}
reservation, err := c.reserveInboundRPCBatch(ctx, specs)
if err != nil {
if errors.Is(err, ErrInboundRPCQueueFull) {
for _, index := range indices {
plan.items[index].kind = inboundItemCapacityError
}
plan.rejectNewRPCOwners(indices)
return nil
}
return err
}
plan.rpcReservation = reservation
plan.rpcTasks = make([]inboundRPC, len(indices))
for i, index := range indices {
item := &plan.items[index]
body := append([]byte(nil), item.body...)
owner, _ := item.payload.(*rpcResultOwnerLease)
plan.rpcTasks[i] = s.newInboundRPCTask(c, item.msgID, specs[i].method, body, owner)
}
return nil
}
func (s *Server) executeInboundPlan(ctx context.Context, cs *connState, c *Conn, plan *inboundPlan) error {
for _, item := range plan.items {
switch item.kind {
case inboundItemDuplicate:
// The preflight plan already stages a content ACK for a locally seen
// duplicate. Do not wait for or replay its result on the same reliable
// stream; the original owner is solely responsible for that response.
continue
case inboundItemServiceDuplicate:
// Classify from the originally committed connState record, never from the
// retransmitted body. This prevents same-id payload replacement. Only an
// already retained answer is eligible for resend (rpc_drop_answer today);
// other best-effort service traffic uses a later fresh request.
if err := s.replayRPCResultByRequest(ctx, c, item.msgID); err != nil {
return err
}
case inboundItemReplayRPC:
if encoded, _ := item.payload.(*encodedOutboundMessage); encoded != nil {
if err := s.sendReplayedRPCResultWithHook(ctx, c, encoded, item.replayAfterSuccessfulDelivery); err != nil {
return err
}
} else if err := s.replayRPCResultByRequest(ctx, c, item.msgID); err != nil {
return err
}
case inboundItemRewrappedRPC:
// Activation is deferred until every session/control barrier has
// committed. It subscribes to the original result event and never waits
// or dispatches the business handler again.
continue
case inboundItemPing:
if err := s.sendPong(ctx, c, item.msgID, item.payload.(mt.PingRequest).PingID); err != nil {
return err
}
case inboundItemPingDelay:
if err := s.sendPong(ctx, c, item.msgID, item.payload.(mt.PingDelayDisconnectRequest).PingID); err != nil {
return err
}
case inboundItemFutureSalts:
if err := s.sendFutureSalts(ctx, c, item.msgID, item.payload.(mt.GetFutureSaltsRequest).Num); err != nil {
return err
}
case inboundItemMsgsAck:
ids := item.payload.(int64VectorView).materialize()
c.AckServerMessages(ids)
s.log.Debug("Received msgs_ack", zap.Int64s("msg_ids", ids))
case inboundItemStateReq:
ids := item.payload.(int64VectorView).materialize()
outgoing, err := c.OutgoingStateInfo(ctx, ids)
if err != nil {
return err
}
if err := s.sendMsgsStateInfo(ctx, c, item.msgID, mergeStateInfo(outgoing, cs.stateInfo(ids))); err != nil {
return err
}
case inboundItemResendReq:
ids := item.payload.(int64VectorView).materialize()
outgoing, err := c.ResendMessages(ctx, ids)
if err != nil {
return err
}
if err := s.sendMsgsStateInfo(ctx, c, item.msgID, mergeStateInfo(outgoing, cs.stateInfo(ids))); err != nil {
return err
}
case inboundItemStateInfo:
value := item.payload.(stateInfoPayload)
s.log.Debug("Received msgs_state_info", zap.Int64("req_msg_id", value.reqMsgID), zap.Int("len", len(value.info)))
case inboundItemAllInfo:
value := item.payload.(allInfoPayload)
s.log.Debug("Received msgs_all_info", zap.Int("msg_ids", value.count), zap.Int("len", len(value.info)))
case inboundItemDestroySession:
if err := s.sendDestroySession(ctx, c, item.payload.(mt.DestroySessionRequest).SessionID); err != nil {
return err
}
case inboundItemHTTPWait:
value := item.payload.(mt.HTTPWaitRequest)
s.log.Debug("Received http_wait", zap.Int("max_delay", value.MaxDelay), zap.Int("wait_after", value.WaitAfter), zap.Int("max_wait", value.MaxWait))
case inboundItemDropAnswer:
value := item.payload.(mt.RPCDropAnswerRequest)
s.log.Debug("Received rpc_drop_answer", zap.Int64("req_msg_id", value.ReqMsgID))
if err := s.sendResult(ctx, c, item.msgID, &mt.RPCAnswerUnknown{}); err != nil {
return err
}
case inboundItemDestroyAuthKey:
s.log.Debug("Received destroy_auth_key", zap.String("auth_key_id", c.authKeyHex))
if err := s.authKeys.Delete(ctx, c.authKeyID); err != nil {
s.log.Warn("Delete auth key failed", zap.String("auth_key_id", c.authKeyHex), zap.Error(err))
return c.SendRequiredControl(ctx, proto.MessageServerResponse, &destroyAuthKeyRPCResult{
RequestMessageID: item.msgID,
ResultTypeID: destroyAuthKeyFailTypeID,
})
}
if registry, ok := s.layerRPC.(LayerRPCSessionProfileRegistry); ok {
registry.ForgetNegotiatedAuthKey(c.authKeyID)
}
// Fence every other active/claiming generation before acknowledging the
// deletion. The exact requester remains writable only long enough to put the
// request-correlated rpc_result(destroy_auth_key_ok) frame on the wire.
s.conns.CloseSessionsForRawAuthKeyExceptConn(c.authKeyID, c)
if err := c.SendRequiredControl(ctx, proto.MessageServerResponse, &destroyAuthKeyRPCResult{
RequestMessageID: item.msgID,
ResultTypeID: destroyAuthKeyOkTypeID,
}); err != nil {
return err
}
// The key cannot reconnect to ACK or replay any old answer. Release every
// logical outbox and receipt only after the terminal OK is physically on
// the wire; retaining them for the offline TTL would be pure leakage.
s.conns.ForgetLogicalSessionsForRawAuthKey(c.authKeyID)
c.beginTerminalShutdown()
c.closeTransport()
return nil
case inboundItemHelpTest:
// Legacy help.test#c0e202f7 = Bool was removed from the published
// schema but is still used by some clients as a probe. Answer with the
// fixed boolTrue terminal; it needs no auth and never reaches routing.
s.log.Debug("Received help.test", zap.Int64("msg_id", item.msgID))
if err := c.SendRequiredControl(ctx, proto.MessageServerResponse, &helpTestRPCResult{
RequestMessageID: item.msgID,
}); err != nil {
return err
}
case inboundItemRPC:
// prepareInboundRPCBatch owns every fresh RPC before synchronous service
// execution begins; commitRPCBatch publishes them after all protocol barriers.
continue
case inboundItemCapacityError:
if owner, _ := item.payload.(*rpcResultOwnerLease); owner != nil {
owner.CompleteExecution(false)
}
if err := s.sendResult(ctx, c, item.msgID, rpcWorkerBusyError()); err != nil {
return err
}
case inboundItemRPCAdmissionError:
rpcErr, _ := item.payload.(*mt.RPCError)
if rpcErr == nil {
rpcErr = &mt.RPCError{ErrorCode: 400, ErrorMessage: "INPUT_REQUEST_INVALID"}
}
if err := s.sendResult(ctx, c, item.msgID, rpcErr); err != nil {
return err
}
default:
return fmt.Errorf("unknown inbound item kind %d", item.kind)
}
}
return nil
}