astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge remote-tracking branch 'upstream/main' into merge-gramsrv-2965f5d

Browse source
This commit is contained in:
onysd 2026-07-20 23:43:51 +03:00
commit ebb0be38d9
355 changed files with 44640 additions and 2320 deletions
Download patch file Download diff file Expand all files Collapse all files

66
internal/mtprotoedge/session_manager.go
View file

@ -1366,15 +1366,19 @@ func (m *SessionManager) PushToUserExceptAuthKeySession(ctx context.Context, use
func (m *SessionManager) PushToUserAuthKey(ctx context.Context, userID int64, businessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass) (int, error) {
// Secret-chat qts is the durable source of truth, so online delivery is an accelerator just
// like account pts fan-out. Do not synchronously wait for every PFS/raw connection's socket.
return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, t, msg, 2*time.Second)
return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, 0, t, msg, 2*time.Second)
}
// PushToUserAuthKeyTransient 是 PushToUserAuthKey 的 transient（typing）best-effort 版本。
func (m *SessionManager) PushToUserAuthKeyTransient(ctx context.Context, userID int64, businessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, t, msg, timeout)
return m.pushToBusinessAuthKeyBestEffort(ctx, userID, businessAuthKeyID, 0, t, msg, timeout)
}
func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, userID int64, businessAuthKeyID [8]byte, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
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) pushToBusinessAuthKeyBestEffort(ctx context.Context, userID int64, businessAuthKeyID [8]byte, minLayer int, t proto.MessageType, msg tg.UpdatesClass, timeout time.Duration) (int, error) {
if ctx != nil && ctx.Err() != nil {
return 0, ctx.Err()
}
@ -1397,7 +1401,7 @@ func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, us
defer cancel()
}
getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
return m.pushToBusinessAuthKey(ctx, userID, businessAuthKeyID, func(c *Conn) error {
return m.pushToBusinessAuthKey(ctx, userID, businessAuthKeyID, minLayer, func(c *Conn) error {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1420,7 +1424,7 @@ func (m *SessionManager) pushToBusinessAuthKeyBestEffort(ctx context.Context, us
})
}
func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64, businessAuthKeyID [8]byte, send func(*Conn) error) (int, error) {
func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64, businessAuthKeyID [8]byte, minLayer int, send func(*Conn) error) (int, error) {
m.mu.Lock()
candidates := m.businessAuthKeyCandidatesLocked(businessAuthKeyID)
conns := make([]*Conn, 0, len(candidates))
@ -1432,6 +1436,9 @@ func (m *SessionManager) pushToBusinessAuthKey(ctx context.Context, userID int64
// 未就绪：密聊消息靠 getDifference 补，typing 直接丢——都不进 pending。
continue
}
if !sessionSupportsMinimumLayer(c, minLayer) {
continue
}
conns = append(conns, c)
}
m.mu.Unlock()
@ -1476,7 +1483,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, t, getUpdates, true, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, 0, t, getUpdates, true, func(c *Conn) error {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1499,7 +1506,25 @@ 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, t, getUpdates, false, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, &excludeAuthKeyID, excludeSessionID, 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) PushToUserTransientAtLeastLayer(ctx context.Context, userID int64, minLayer int, 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 {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1520,6 +1545,10 @@ 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()
}
@ -1545,7 +1574,7 @@ func (m *SessionManager) pushToUserBestEffort(ctx context.Context, userID int64,
defer cancel()
}
getUpdates := onceLayerUpdatesFanout(sendCtx, msg)
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, t, getUpdates, true, func(c *Conn) error {
return m.pushToUserWithSender(ctx, userID, excludeAuthKeyID, excludeSessionID, minLayer, t, getUpdates, true, func(c *Conn) error {
if c.outbound == nil || c.outboundControl == nil {
return ErrConnClosed
}
@ -1592,7 +1621,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, 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, minLayer int, 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)
@ -1612,6 +1641,10 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
excluded++
continue
}
if !sessionSupportsMinimumLayer(c, minLayer) {
skipped++
continue
}
if !c.receivesUpdates.Load() {
if !queueWhenNotReady {
// transient（typing/presence）：未就绪即丢，不进 pending。这些 update 不写
@ -1640,6 +1673,10 @@ func (m *SessionManager) pushToUserWithSender(ctx context.Context, userID int64,
excluded++
continue
}
if !sessionSupportsMinimumLayer(c, minLayer) {
skipped++
continue
}
if !c.receivesUpdates.Load() {
if !queueWhenNotReady {
skipped++
@ -2478,6 +2515,17 @@ func shouldExcludeSession(c *Conn, excludeAuthKeyID *[8]byte, excludeSessionID i
return c.authKeyID == *excludeAuthKeyID
}
func sessionSupportsMinimumLayer(c *Conn, minLayer int) bool {
if minLayer <= 0 {
return true
}
if c == nil {
return false
}
state := c.LayerProfileState()
return state.Origin != LayerProfileUnknown && int(state.Profile) >= minLayer
}
func sessionKeyLog(id [8]byte) string {
return fmt.Sprintf("%x", id)
}

56
internal/mtprotoedge/transient_push_test.go
View file

@ -3,11 +3,13 @@ package mtprotoedge
import (
"context"
"testing"
"time"
"go.uber.org/zap/zaptest"
"github.com/iamxvbaba/td/proto"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tlprofile"
)
// TestPushTransientSkipsNotReadySession 锁定不变量：transient 推送（typing/presence）对
@ -51,3 +53,57 @@ func TestPushTransientSkipsNotReadySession(t *testing.T) {
t.Fatalf("durable push queued %d pending, want 1", n)
}
}
// 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) {
sm := NewSessionManager(zaptest.NewLogger(t))
const userID = int64(101)
makeConn := func(sessionID int64, profile tlprofile.Profile, known bool) *Conn {
c := &Conn{
sessionID: sessionID, authKeyID: [8]byte{byte(sessionID)},
outbound: make(chan outboundOp, 2), outboundControl: make(chan outboundOp, 2),
outboundStop: make(chan struct{}),
}
c.userID.Store(userID)
c.userIDResolved.Store(true)
c.receivesUpdates.Store(true)
if known {
if err := c.FreezeLayerProfile(profile); err != nil {
t.Fatal(err)
}
}
if err := sm.Register(c); err != nil {
t.Fatal(err)
}
return c
}
old := makeConn(1, tlprofile.Profile227, true)
current := makeConn(2, tlprofile.Profile228, true)
unknown := makeConn(3, 0, false)
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 {
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 old.isRetired() || unknown.isRetired() {
t.Fatal("unsupported transient update retired an old/unknown session")
}
for _, c := range []*Conn{old, current, unknown} {
sm.mu.RLock()
pending := len(sm.pending[connSessionKey(c)])
sm.mu.RUnlock()
if pending != 0 {
t.Fatalf("session %d queued %d transient updates", c.sessionID, pending)
}
}
}