perf(imap): STORE 批量持久化 + 推送序号复用已加载列表,消除逐条写库与全量扫描

原实现每封匹配消息单独 UPDATE + GetByID + seqOf 全量扫描(含大
附件 raw_data),手机整批标记已读(60+ 封)时产生 60 次写 +
180 次全量读,连接被长时间占住,其他连接响应被推送洪泛阻塞。

优化:收集目标状态后合并为单条 UPDATE ... IN 批量写;推送更新
直接用已加载列表的序号与目标状态(buildFlagsUpdateAt),不再
重复查库。实测 19 封批量 STORE 服务器耗时 372µs。
This commit is contained in:
dsh
2026-08-19 11:13:42 -04:00
parent 9fef9da830
commit aa0437fb4f
2 changed files with 126 additions and 38 deletions
+104 -38
View File
@@ -83,10 +83,20 @@ func buildFlagsUpdate(stores *store.Stores, userEmail, mailbox string, msg *db.M
if stores == nil || msg == nil || userEmail == "" || mailbox == "" { if stores == nil || msg == nil || userEmail == "" || mailbox == "" {
return nil return nil
} }
imapMsg := imap.NewMessage(seqOf(stores, msg.UserID, mailbox, msg.ID), return buildFlagsUpdateAt(seqOf(stores, msg.UserID, mailbox, msg.ID), userEmail, mailbox, msg, msg.IsRead, msg.IsFlagged, deleted)
}
// buildFlagsUpdateAt 与 buildFlagsUpdate 相同,但序号与已读/星标状态由
// 调用方直接提供(批量 STORE 路径已加载全量列表,避免逐条 GetByID +
// seqOf 全量扫描)。
func buildFlagsUpdateAt(seq uint32, userEmail, mailbox string, msg *db.Message, read, flagged, deleted bool) *backend.MessageUpdate {
if msg == nil || userEmail == "" || mailbox == "" {
return nil
}
imapMsg := imap.NewMessage(seq,
[]imap.FetchItem{imap.FetchUid, imap.FetchFlags}) []imap.FetchItem{imap.FetchUid, imap.FetchFlags})
imapMsg.Uid = uint32(msg.ID) imapMsg.Uid = uint32(msg.ID)
imapMsg.Flags = flagsOf(msg.IsRead, msg.IsFlagged, deleted) imapMsg.Flags = flagsOf(read, flagged, deleted)
return &backend.MessageUpdate{ return &backend.MessageUpdate{
Update: backend.NewUpdate(userEmail, mailbox), Update: backend.NewUpdate(userEmail, mailbox),
Message: imapMsg, Message: imapMsg,
@@ -790,12 +800,30 @@ func (m *imapMailbox) UpdateMessagesFlags(uid bool, seqset *imap.SeqSet, op imap
if err != nil { if err != nil {
return err return err
} }
// 批量模式:先收集每封匹配消息的目标标志状态,再合并成少量 SQL
// 写入。此前逐条 UPDATE + 逐条 GetByID + 逐条 seqOf 全量扫描,
// 手机整批标记已读(60+ 封)时会产生 60 次写 + 180 次全量读
// (含大附件 raw_data),该连接命令循环被长时间占住,其他连接
// 的响应通道也被推送洪泛阻塞——表现为"卡在接收邮件"。
flagSet := make(map[string]bool, len(flags))
for _, flag := range flags {
flagSet[flag] = true
}
// 记录首个持久化错误:SQLite 忙/锁等瞬时失败必须让客户端感知 type change struct {
// (返回 NO 触发重试),否则已读/星标会静默丢失。 msg *db.Message
var firstErr error seq uint32
newRead bool
readSet bool
newFlagged bool
flaggedSet bool
newDeleted bool
deletedSet bool
}
var changes []change
for i, dbMsg := range dbMessages { for i := range dbMessages {
dbMsg := &dbMessages[i]
var match bool var match bool
if uid { if uid {
match = seqset.Contains(uint32(dbMsg.ID)) match = seqset.Contains(uint32(dbMsg.ID))
@@ -806,55 +834,93 @@ func (m *imapMailbox) UpdateMessagesFlags(uid bool, seqset *imap.SeqSet, op imap
continue continue
} }
flagSet := make(map[string]bool, len(flags)) c := change{msg: dbMsg, seq: uint32(i + 1)}
for _, flag := range flags { apply := func(flag string, enabled bool) {
flagSet[flag] = true
}
applyFlag := func(flag string, enabled bool) {
switch flag { switch flag {
case "\\Seen": case "\\Seen":
if err := m.stores.Mails.MarkReadState(dbMsg.ID, enabled); err != nil && firstErr == nil { c.newRead, c.readSet = enabled, true
log.Printf("IMAP: mark read state for msg %d failed: %v", dbMsg.ID, err)
firstErr = err
}
case "\\Flagged": case "\\Flagged":
if err := m.stores.Mails.MarkFlagged(dbMsg.ID, enabled); err != nil && firstErr == nil { c.newFlagged, c.flaggedSet = enabled, true
log.Printf("IMAP: mark flagged for msg %d failed: %v", dbMsg.ID, err)
firstErr = err
}
case "\\Deleted": case "\\Deleted":
if enabled { c.newDeleted, c.deletedSet = enabled, true
m.deleted[dbMsg.ID] = true
} else {
delete(m.deleted, dbMsg.ID)
} }
} }
}
switch op { switch op {
case imap.SetFlags: case imap.SetFlags:
applyFlag("\\Seen", flagSet["\\Seen"]) apply("\\Seen", flagSet["\\Seen"])
applyFlag("\\Flagged", flagSet["\\Flagged"]) apply("\\Flagged", flagSet["\\Flagged"])
applyFlag("\\Deleted", flagSet["\\Deleted"]) apply("\\Deleted", flagSet["\\Deleted"])
case imap.AddFlags: case imap.AddFlags:
for flag := range flagSet { for flag := range flagSet {
applyFlag(flag, true) apply(flag, true)
} }
case imap.RemoveFlags: case imap.RemoveFlags:
for flag := range flagSet { for flag := range flagSet {
applyFlag(flag, false) apply(flag, false)
} }
} }
changes = append(changes, c)
}
// 标志变化(已读/星标/删除)→ 推送给同用户其他客户端 // 批量持久化(已读/星标合并为单条 UPDATE ... IN
// (重新读库取最新状态,\Deleted 取会话内状态) var readTrue, readFalse, flagTrue, flagFalse []uint
fresh, err := m.stores.Mails.GetByID(dbMsg.ID) for _, c := range changes {
if err != nil { if c.readSet && c.newRead != c.msg.IsRead {
continue if c.newRead {
readTrue = append(readTrue, c.msg.ID)
} else {
readFalse = append(readFalse, c.msg.ID)
} }
deleted := m.deleted != nil && m.deleted[dbMsg.ID] }
pushUpdate(m.user.updates, buildFlagsUpdate(m.stores, m.user.email, m.name, fresh, deleted)) if c.flaggedSet && c.newFlagged != c.msg.IsFlagged {
if c.newFlagged {
flagTrue = append(flagTrue, c.msg.ID)
} else {
flagFalse = append(flagFalse, c.msg.ID)
}
}
}
// 记录首个持久化错误:SQLite 忙/锁等瞬时失败必须让客户端感知
// (返回 NO 触发重试),否则已读/星标会静默丢失。
var firstErr error
mark := func(err error) {
if err != nil && firstErr == nil {
firstErr = err
}
}
if len(readTrue) > 0 {
mark(m.stores.Mails.SetReadStates(readTrue, true))
}
if len(readFalse) > 0 {
mark(m.stores.Mails.SetReadStates(readFalse, false))
}
if len(flagTrue) > 0 {
mark(m.stores.Mails.SetFlaggedStates(flagTrue, true))
}
if len(flagFalse) > 0 {
mark(m.stores.Mails.SetFlaggedStates(flagFalse, false))
}
// 标志变化 → 推送给同用户其他客户端(状态取目标值,序号用已加载
// 列表的下标——与 ListMessages/Status 全链路一致,不再重复全量扫描)
for _, c := range changes {
if c.deletedSet {
if c.newDeleted {
m.deleted[c.msg.ID] = true
} else {
delete(m.deleted, c.msg.ID)
}
}
read := c.msg.IsRead
if c.readSet {
read = c.newRead
}
flagged := c.msg.IsFlagged
if c.flaggedSet {
flagged = c.newFlagged
}
deleted := m.deleted != nil && m.deleted[c.msg.ID]
pushUpdate(m.user.updates, buildFlagsUpdateAt(c.seq, m.user.email, m.name, c.msg, read, flagged, deleted))
} }
return firstErr return firstErr
+22
View File
@@ -28,6 +28,10 @@ type MailStore interface {
MarkRead(id uint) error MarkRead(id uint) error
MarkReadState(id uint, read bool) error MarkReadState(id uint, read bool) error
MarkFlagged(id uint, flagged bool) error MarkFlagged(id uint, flagged bool) error
// SetReadStates 批量设置多封邮件的已读状态(单条 UPDATE ... IN)。
SetReadStates(ids []uint, read bool) error
// SetFlaggedStates 批量设置多封邮件的星标状态(单条 UPDATE ... IN)。
SetFlaggedStates(ids []uint, flagged bool) error
MoveToFolder(id uint, folder string) error MoveToFolder(id uint, folder string) error
Delete(id uint) error Delete(id uint) error
CountUnread(userID uint, folder string) (int64, error) CountUnread(userID uint, folder string) (int64, error)
@@ -103,6 +107,24 @@ func (s *mailStoreGorm) MarkFlagged(id uint, flagged bool) error {
return s.db.Model(&db.Message{}).Where("id = ?", id).Update("is_flagged", flagged).Error return s.db.Model(&db.Message{}).Where("id = ?", id).Update("is_flagged", flagged).Error
} }
// SetReadStates 批量设置多封邮件的已读状态。
// 客户端整批标记已读(手机同步后 STORE +FLAGS \Seen)时,逐条 UPDATE
// 会产生大量写事务并占住连接,这里合并为单条 SQL。
func (s *mailStoreGorm) SetReadStates(ids []uint, read bool) error {
if len(ids) == 0 {
return nil
}
return s.db.Model(&db.Message{}).Where("id IN ?", ids).Update("is_read", read).Error
}
// SetFlaggedStates 批量设置多封邮件的星标状态。
func (s *mailStoreGorm) SetFlaggedStates(ids []uint, flagged bool) error {
if len(ids) == 0 {
return nil
}
return s.db.Model(&db.Message{}).Where("id IN ?", ids).Update("is_flagged", flagged).Error
}
// MoveToFolder changes the folder of a message. // MoveToFolder changes the folder of a message.
func (s *mailStoreGorm) MoveToFolder(id uint, folder string) error { func (s *mailStoreGorm) MoveToFolder(id uint, folder string) error {
return s.db.Model(&db.Message{}).Where("id = ?", id).Update("folder", folder).Error return s.db.Model(&db.Message{}).Where("id = ?", id).Update("folder", folder).Error