diff --git a/README.md b/README.md index 3cfa792..f1fbb6f 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Web 前端采用 QQ 邮箱风格的布局：顶部导航 + 左侧文件夹栏 + ## 功能特性 - **邮件协议**：SMTP（发送）、IMAP（同步）、POP3（收取），均支持 TLS 加密 -- **外部投递**：认证用户可向外部邮箱（QQ/Gmail/Outlook 等）发送邮件，内置外发队列、MX 直投、STARTTLS、指数退避重试、退信通知与 DKIM 签名 +- **外部投递**：认证用户可向外部邮箱（QQ/Gmail/Outlook 等）发送邮件，内置外发队列、**并发 worker 池投递**（默认 4 线程 + 每收件域并发上限 2）、MX 直投、STARTTLS、指数退避重试、退信通知与 DKIM 签名 - **Web 邮箱**：QQ 邮箱风格界面，支持收件箱 / 已发送 / 草稿箱、未读角标与搜索过滤、全选 / 批量删除、发件人头像、富文本编辑（Quill.js）、附件上传/下载 - **管理后台**：域名管理、用户管理、DKIM 密钥自动生成、DNS 配置提示、全量邮件查看、外发队列管理、IP 封禁管理、仪表盘统计 - **协议调用日志**：SMTP / IMAP / POP3 每次连接自动记录来源 IP、用户名、成功/失败、失败原因与操作摘要，可按协议/状态/IP/用户名/时间筛选，用于分析密码爆破、中继滥用等攻击行为（默认保留 30 天，自动清理） @@ -139,6 +139,11 @@ data_dir = "" # Caddy 数据目录（含 certificate [outbound] hostname = "" # EHLO 主机名，留空使用 [smtp] domain poll_interval = 15 # 外发队列扫描间隔（秒） +workers = 4 # 并发投递 worker 数（多线程并行发送， + # 大量邮件时吞吐提升；0/1 为串行） +batch_size = 50 # 每次扫描最多取出的待投递邮件数 +max_concurrent_per_domain = 2 # 同一收件域（或中继）的最大并发连接数， + # 防被判定为滥发；0 表示不限制 max_attempts = 12 # 单封邮件最大投递尝试次数 retry_base_min = 5 # 重试退避基数（分钟），指数增长：5/10/20/40... max_recipients = 50 # 单封邮件最大外部收件人数 diff --git a/config/config.go b/config/config.go index 1e79c38..9125626 100644 --- a/config/config.go +++ b/config/config.go @@ -126,6 +126,15 @@ type OutboundConfig struct { MaxPerDay int `toml:"max_per_day"` // 每用户每日最大外发数，0 表示禁用外部投递 ConnectTimeout int `toml:"connect_timeout"` // 连接远程 MX 超时（秒） + // Workers 并发投递 worker 数：多 goroutine 并行发送队列中的邮件。 + // 0 或 1 表示串行（旧行为）。 + Workers int `toml:"workers"` + // BatchSize 每次扫描最多取出的待投递邮件数。 + BatchSize int `toml:"batch_size"` + // MaxConcurrentPerDomain 同一收件域（或中继）的最大并发连接数， + // 防止对单个 MX 域并发过多而被判定为滥发；0 表示不限制。 + MaxConcurrentPerDomain int `toml:"max_concurrent_per_domain"` + // Smarthost relay: when relay_host is non-empty, all external mail is // delivered through this relay instead of direct MX delivery. Useful when // the server IP is listed in PBL/blocklists (residential/dynamic IPs). @@ -230,16 +239,19 @@ func defaultConfig() *Config { // Caddy: 留空则自动探测常见数据目录，无需配置 Caddy: CaddyConfig{}, Outbound: OutboundConfig{ - PollInterval: 15, // 15 秒扫描一次队列 - MaxAttempts: 12, // 最多尝试 12 次 - RetryBaseMin: 5, // 5/10/20/40/... 分钟指数退避 - MaxRecipients: 50, // 单封最多 50 个外部收件人 - MaxPerMin: 30, // 每用户每分钟 30 封 - MaxPerDay: 500, - ConnectTimeout: 30, // 连接远程 MX 超时 30 秒 - RelayPort: 587, // smarthost 默认提交端口 - RelayStartTLS: true, - IPFamily: "ipv4", + PollInterval: 15, // 15 秒扫描一次队列 + MaxAttempts: 12, // 最多尝试 12 次 + RetryBaseMin: 5, // 5/10/20/40/... 分钟指数退避 + MaxRecipients: 50, // 单封最多 50 个外部收件人 + MaxPerMin: 30, // 每用户每分钟 30 封 + MaxPerDay: 500, + ConnectTimeout: 30, // 连接远程 MX 超时 30 秒 + RelayPort: 587, // smarthost 默认提交端口 + RelayStartTLS: true, + IPFamily: "ipv4", + Workers: DefaultOutboundWorkers, + BatchSize: DefaultOutboundBatchSize, + MaxConcurrentPerDomain: DefaultMaxConcurrentPerDomain, }, } } @@ -325,6 +337,15 @@ func mergeDefaults(cfg *Config, defaults *Config) *Config { if cfg.Outbound.ConnectTimeout == 0 { cfg.Outbound.ConnectTimeout = defaults.Outbound.ConnectTimeout } + if cfg.Outbound.Workers == 0 { + cfg.Outbound.Workers = defaults.Outbound.Workers + } + if cfg.Outbound.BatchSize == 0 { + cfg.Outbound.BatchSize = defaults.Outbound.BatchSize + } + if cfg.Outbound.MaxConcurrentPerDomain == 0 { + cfg.Outbound.MaxConcurrentPerDomain = defaults.Outbound.MaxConcurrentPerDomain + } if cfg.Outbound.RelayPort == 0 { cfg.Outbound.RelayPort = defaults.Outbound.RelayPort } diff --git a/config/defaults.go b/config/defaults.go index 0e3dbdd..5cefac8 100644 --- a/config/defaults.go +++ b/config/defaults.go @@ -39,5 +39,15 @@ const ( // DefaultProtocolLogKeepDays 是 SMTP/IMAP/POP3 协议调用日志的默认保留天数。 const DefaultProtocolLogKeepDays = 30 +// Outbound delivery concurrency defaults. +const ( + // DefaultOutboundWorkers 并发投递 worker 数（0/1 为串行）。 + DefaultOutboundWorkers = 4 + // DefaultOutboundBatchSize 每次扫描最多取出的待投递邮件数。 + DefaultOutboundBatchSize = 50 + // DefaultMaxConcurrentPerDomain 同一收件域的最大并发连接数。 + DefaultMaxConcurrentPerDomain = 2 +) + // ConfigFileName is the name of the configuration file const ConfigFileName = "mail_go.toml" diff --git a/internal/outbound/manager.go b/internal/outbound/manager.go index 5957412..00cccef 100644 --- a/internal/outbound/manager.go +++ b/internal/outbound/manager.go @@ -16,8 +16,13 @@ import ( ) // Manager orchestrates the outbound delivery queue: enqueueing messages, -// background delivery worker, exponential backoff retries, DKIM signing, -// per-user rate limits and failure bounces. +// concurrent background delivery workers, exponential backoff retries, DKIM +// signing, per-user rate limits and failure bounces. +// +// 并发模型：一个 dispatcher goroutine 周期性扫描队列并原子抢占（Claim） +// 待投递项，投递给 worker 池（默认 4 个 goroutine）并行发送；同一收件域 +// （或中继）的连接数受 max_concurrent_per_domain 限制。workers=0/1 时退 +// 化为串行投递（旧行为）。 type Manager struct { cfg config.OutboundConfig hostname string // EHLO hostname @@ -31,7 +36,12 @@ type Manager struct { wg sync.WaitGroup mu sync.Mutex lim map[uint]*userWindow - batch int + jobs chan *db.OutboundMessage + domMu sync.Mutex + dom map[string]chan struct{} // 每域并发信号量 + + // deliver 执行单封投递，默认走 m.mailer.Deliver；测试可注入替换。 + deliver func(from, to string, data []byte) (string, error) } // userWindow tracks a user's sending rate within fixed windows. @@ -45,6 +55,14 @@ type userWindow struct { // NewManager creates an outbound delivery Manager. // hostname is the EHLO name presented to remote servers (defaults to "localhost"). func NewManager(cfg config.OutboundConfig, hostname string, stores *store.Stores) *Manager { + batchSize := cfg.BatchSize + if batchSize <= 0 { + batchSize = 50 + } + workers := cfg.Workers + if workers <= 1 { + workers = 1 + } m := &Manager{ cfg: cfg, hostname: hostname, @@ -54,8 +72,10 @@ func NewManager(cfg config.OutboundConfig, hostname string, stores *store.Stores stop: make(chan struct{}), done: make(chan struct{}), lim: make(map[uint]*userWindow), - batch: 50, + jobs: make(chan *db.OutboundMessage, batchSize), + dom: make(map[string]chan struct{}), } + m.deliver = m.mailer.Deliver m.mailer.IPFamily = cfg.IPFamily m.mailer.SourceIP = cfg.SourceIP if cfg.SourceIP != "" { @@ -73,22 +93,27 @@ func NewManager(cfg config.OutboundConfig, hostname string, stores *store.Stores } log.Printf("outbound: using smarthost relay %s:%d", cfg.RelayHost, cfg.RelayPort) } + log.Printf("outbound: %d delivery workers, batch=%d, per-domain concurrency=%d", + workers, batchSize, cfg.MaxConcurrentPerDomain) return m } -// Start launches the background delivery worker. +// Start launches the dispatcher and the delivery worker pool. func (m *Manager) Start() { interval := time.Duration(m.cfg.PollInterval) * time.Second if interval <= 0 { interval = 15 * time.Second } + // 调度者：启动时立即扫描一次（清空积压），之后按周期扫描 + + // 原子抢占待投递项，投递给 worker 池。 m.wg.Add(1) go func() { defer m.wg.Done() ticker := time.NewTicker(interval) defer ticker.Stop() - log.Printf("outbound: delivery worker started (interval=%s, max_attempts=%d)", interval, m.cfg.MaxAttempts) + log.Printf("outbound: dispatcher started (interval=%s, max_attempts=%d)", interval, m.cfg.MaxAttempts) + m.processDue() for { select { case <-ticker.C: @@ -96,14 +121,30 @@ func (m *Manager) Start() { case <-m.kick: m.processDue() case <-m.stop: + close(m.jobs) close(m.done) return } } }() + + // Worker 池：并行投递。 + workers := m.cfg.Workers + if workers <= 1 { + workers = 1 + } + for i := 0; i < workers; i++ { + m.wg.Add(1) + go func() { + defer m.wg.Done() + for job := range m.jobs { + m.deliverOne(job) + } + }() + } } -// Stop gracefully stops the delivery worker. +// Stop gracefully stops the dispatcher and workers. func (m *Manager) Stop() { m.once.Do(func() { close(m.stop) @@ -243,28 +284,77 @@ func (m *Manager) checkRateLimit(userID uint) error { return nil } -// processDue attempts delivery of all due queue items. +// processDue scans the queue and dispatches due items to the worker pool. +// Each item is atomically claimed (status -> sending) before dispatch so +// that concurrent workers never deliver the same message twice. When all +// workers are busy the dispatcher blocks here, naturally throttling claims; +// remaining due items are picked up on the next scan. func (m *Manager) processDue() { - items, err := m.stores.Outbound.ListDue(time.Now(), m.batch) + batchSize := m.cfg.BatchSize + if batchSize <= 0 { + batchSize = 50 + } + items, err := m.stores.Outbound.ListDue(time.Now(), batchSize) if err != nil { log.Printf("outbound: loading due queue failed: %v", err) return } for i := range items { - m.deliverOne(&items[i]) + claimed, err := m.stores.Outbound.Claim(items[i].ID) + if err != nil { + log.Printf("outbound: claim item %d failed: %v", items[i].ID, err) + continue + } + if !claimed { + // 已被其他调度周期抢占（并发下应不会发生，防御性跳过） + continue + } + item := items[i] + item.Status = db.OutboundStatusSending + m.jobs <- &item } } -// deliverOne performs a single delivery attempt for a queue item. -func (m *Manager) deliverOne(item *db.OutboundMessage) { - // Mark as sending to avoid concurrent workers double-delivering. - item.Status = db.OutboundStatusSending - if err := m.stores.Outbound.Update(item); err != nil { - log.Printf("outbound: update item %d to sending failed: %v", item.ID, err) - return +// acquireDomain 获取收件域（或中继）的并发信号量，限制对同一目标域同时 +// 打开的 SMTP 连接数。limit <= 0 表示不限制。 +func (m *Manager) acquireDomain(domain string) func() { + limit := m.cfg.MaxConcurrentPerDomain + if limit <= 0 { + return func() {} } - resp, err := m.mailer.Deliver(item.FromAddr, item.ToAddr, []byte(item.RawData)) + m.domMu.Lock() + sem := m.dom[domain] + if sem == nil { + sem = make(chan struct{}, limit) + m.dom[domain] = sem + } + m.domMu.Unlock() + + sem <- struct{}{} + return func() { <-sem } +} + +// deliverKey 返回并发限制使用的目标标识：配置了中继时所有连接都打向同一 +// smarthost，统一按 "relay" 限制；否则按收件域名限制。 +func (m *Manager) deliverKey(to string) string { + if m.mailer.Relay != nil && m.mailer.Relay.Host != "" { + return "relay" + } + at := strings.LastIndex(to, "@") + if at < 0 || at == len(to)-1 { + return "" + } + return strings.ToLower(to[at+1:]) +} + +// deliverOne performs a single delivery attempt for a queue item. +// 调用前该项已被原子抢占为 sending，此处不再重复置位。 +func (m *Manager) deliverOne(item *db.OutboundMessage) { + release := m.acquireDomain(item.ToAddr) + defer release() + + resp, err := m.deliver(item.FromAddr, item.ToAddr, []byte(item.RawData)) now := time.Now() item.Attempts++ diff --git a/internal/outbound/manager_test.go b/internal/outbound/manager_test.go new file mode 100644 index 0000000..b629319 --- /dev/null +++ b/internal/outbound/manager_test.go @@ -0,0 +1,332 @@ +package outbound + +import ( + "errors" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "mail_go/config" + "mail_go/internal/db" + "mail_go/internal/store" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +// newTestManagerStores 创建带 outbound_messages 表的测试数据库。 +func newTestManagerStores(t *testing.T) *store.Stores { + t.Helper() + gdb, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "test.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := gdb.AutoMigrate(&db.OutboundMessage{}); err != nil { + t.Fatalf("migrate: %v", err) + } + return store.NewStores(gdb) +} + +// seedPending 写入 n 条立即可投递的 pending 队列项。 +func seedPending(t *testing.T, stores *store.Stores, n int) []uint { + t.Helper() + ids := make([]uint, 0, n) + for i := 0; i < n; i++ { + item := &db.OutboundMessage{ + MessageID: "", + FromAddr: "sender@test.local", + ToAddr: "rcpt@fake.test", + RecipientDom: "fake.test", + RawData: "From: sender@test.local\r\nTo: rcpt@fake.test\r\nSubject: t\r\n\r\nbody\r\n", + Status: db.OutboundStatusPending, + Attempts: 0, + NextAttemptAt: time.Now(), + } + if err := stores.Outbound.Create(item); err != nil { + t.Fatalf("create item: %v", err) + } + ids = append(ids, item.ID) + } + return ids +} + +// countByStatus 统计队列中指定状态的项数。 +func countByStatus(t *testing.T, stores *store.Stores, status string) int64 { + t.Helper() + n, err := stores.Outbound.CountByStatus(status) + if err != nil { + t.Fatalf("count %s: %v", status, err) + } + return n +} + +// waitFor 轮询等待条件满足或超时。 +func waitFor(t *testing.T, timeout time.Duration, desc string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", desc) +} + +// concurrencyTracker 统计并发调用数（用于验证 worker 池与每域信号量）。 +type concurrencyTracker struct { + mu sync.Mutex + active int + peak int + deliveries int + err error +} + +// deliverWithDelay 返回一个带固定延迟的注入投递函数，并统计并发峰值。 +func deliverWithDelay(tr *concurrencyTracker, delay time.Duration) func(string, string, []byte) (string, error) { + return func(from, to string, data []byte) (string, error) { + tr.mu.Lock() + tr.active++ + if tr.active > tr.peak { + tr.peak = tr.active + } + tr.mu.Unlock() + + time.Sleep(delay) + + tr.mu.Lock() + tr.active-- + tr.deliveries++ + tr.mu.Unlock() + return "250 2.0.0 queued", nil + } +} + +// TestManagerConcurrentDelivery 验证 worker 池真并发投递且每封只投一次。 +func TestManagerConcurrentDelivery(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 12) + + tr := &concurrencyTracker{} + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 5, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 4, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = deliverWithDelay(tr, 150*time.Millisecond) + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "all items sent", func() bool { + return countByStatus(t, stores, db.OutboundStatusSent) == 12 + }) + + if tr.peak < 2 { + t.Fatalf("expected concurrent deliveries (peak=%d), got serial behavior", tr.peak) + } + if tr.deliveries != 12 { + t.Fatalf("deliveries = %d, want 12 (each message exactly once)", tr.deliveries) + } + if n := countByStatus(t, stores, db.OutboundStatusPending) + countByStatus(t, stores, db.OutboundStatusDeferred); n != 0 { + t.Fatalf("%d items still pending/deferred", n) + } +} + +// TestManagerSerialFallback 验证 workers=1 时退化为串行（旧行为）。 +func TestManagerSerialFallback(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 6) + + tr := &concurrencyTracker{} + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 5, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 1, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = deliverWithDelay(tr, 50*time.Millisecond) + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "all items sent", func() bool { + return countByStatus(t, stores, db.OutboundStatusSent) == 6 + }) + if tr.peak > 1 { + t.Fatalf("workers=1 must be serial, peak=%d", tr.peak) + } +} + +// TestManagerDomainLimit 验证同一收件域的并发连接数不超过上限。 +func TestManagerDomainLimit(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 8) + + tr := &concurrencyTracker{} + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 5, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 8, + BatchSize: 50, + MaxConcurrentPerDomain: 1, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = deliverWithDelay(tr, 100*time.Millisecond) + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "all items sent", func() bool { + return countByStatus(t, stores, db.OutboundStatusSent) == 8 + }) + if tr.peak > 1 { + t.Fatalf("per-domain limit 1 violated: peak=%d", tr.peak) + } + if tr.deliveries != 8 { + t.Fatalf("deliveries = %d, want 8", tr.deliveries) + } +} + +// TestManagerRetriesTemporaryFailure 验证临时失败进入退避重试（deferred）。 +func TestManagerRetriesTemporaryFailure(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 1) + + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 3, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 2, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = func(from, to string, data []byte) (string, error) { + return "", newTempError("connection refused") + } + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "item deferred", func() bool { + return countByStatus(t, stores, db.OutboundStatusDeferred) == 1 + }) + + items, _, err := stores.Outbound.List(1, 10, db.OutboundStatusDeferred) + if err != nil || len(items) != 1 { + t.Fatalf("list deferred: %v (n=%d)", err, len(items)) + } + if items[0].Attempts != 1 { + t.Fatalf("attempts = %d, want 1", items[0].Attempts) + } + if items[0].LastError == "" { + t.Fatal("expected last error recorded") + } +} + +// TestManagerPermanentFailureBouncesAndFails 验证永久失败直接标记 failed。 +func TestManagerPermanentFailureBouncesAndFails(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 1) + + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 3, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 2, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = func(from, to string, data []byte) (string, error) { + return "", newPermError("550 recipient rejected") + } + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "item failed", func() bool { + return countByStatus(t, stores, db.OutboundStatusFailed) == 1 + }) + if n := countByStatus(t, stores, db.OutboundStatusDeferred); n != 0 { + t.Fatalf("permanent failure must not defer: %d deferred", n) + } +} + +// TestManagerDomainLimitNoLimit 验证 max_concurrent_per_domain=0 不限制并发。 +func TestManagerDomainLimitNoLimit(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 8) + + tr := &concurrencyTracker{} + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 5, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 8, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = deliverWithDelay(tr, 80*time.Millisecond) + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "all items sent", func() bool { + return countByStatus(t, stores, db.OutboundStatusSent) == 8 + }) + if tr.peak < 2 { + t.Fatalf("expected concurrent deliveries without domain limit, peak=%d", tr.peak) + } +} + +// TestManagerDeliverErrorsPropagate 防御：投递函数报错不影响 worker 存活， +// 错误项进入重试（deferred）。 +func TestManagerDeliverErrorsPropagate(t *testing.T) { + stores := newTestManagerStores(t) + seedPending(t, stores, 3) + + var calls atomic.Int32 + cfg := config.OutboundConfig{ + PollInterval: 1, + MaxAttempts: 2, + RetryBaseMin: 1, + MaxPerDay: 10000, + Workers: 2, + BatchSize: 50, + ConnectTimeout: 10, + } + m := NewManager(cfg, "test.local", stores) + m.deliver = func(from, to string, data []byte) (string, error) { + n := calls.Add(1) + if n%2 == 0 { + return "", errors.New("boom") + } + return "250 ok", nil + } + m.Start() + t.Cleanup(m.Stop) + + waitFor(t, 15*time.Second, "queue settled", func() bool { + return countByStatus(t, stores, db.OutboundStatusSent)+countByStatus(t, stores, db.OutboundStatusDeferred) == 3 + }) + if calls.Load() != 3 { + t.Fatalf("deliver calls = %d, want 3 (once per item)", calls.Load()) + } + if n := countByStatus(t, stores, db.OutboundStatusFailed); n != 0 { + t.Fatalf("unexpected failed items: %d", n) + } +} diff --git a/internal/store/outbound_store.go b/internal/store/outbound_store.go index 6922089..a902f91 100644 --- a/internal/store/outbound_store.go +++ b/internal/store/outbound_store.go @@ -15,6 +15,9 @@ type OutboundStore interface { ListDue(now time.Time, limit int) ([]db.OutboundMessage, error) List(page, size int, status string) ([]db.OutboundMessage, int64, error) Update(msg *db.OutboundMessage) error + // Claim 原子地将一项待投递邮件置为 sending；仅当该项仍处于 + // pending/deferred 时成功（并发 worker 抢占，防重复投递）。 + Claim(id uint) (bool, error) Delete(id uint) error CountByStatus(status string) (int64, error) } @@ -85,6 +88,15 @@ func (s *outboundStoreGorm) Update(msg *db.OutboundMessage) error { return s.db.Save(msg).Error } +// Claim 原子抢占：把 pending/deferred 项置为 sending。 +// 返回是否抢占成功（false 表示已被其他 worker 抢先或状态已变化）。 +func (s *outboundStoreGorm) Claim(id uint) (bool, error) { + res := s.db.Model(&db.OutboundMessage{}). + Where("id = ? AND status IN (?, ?)", id, db.OutboundStatusPending, db.OutboundStatusDeferred). + Update("status", db.OutboundStatusSending) + return res.RowsAffected == 1, res.Error +} + // Delete removes an outbound queue record by ID. func (s *outboundStoreGorm) Delete(id uint) error { return s.db.Delete(&db.OutboundMessage{}, id).Error diff --git a/internal/store/outbound_store_test.go b/internal/store/outbound_store_test.go new file mode 100644 index 0000000..c3833c0 --- /dev/null +++ b/internal/store/outbound_store_test.go @@ -0,0 +1,102 @@ +package store + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "mail_go/internal/db" +) + +// TestOutboundClaimAtomic 验证并发抢占同一队列项恰好只有一次成功。 +func TestOutboundClaimAtomic(t *testing.T) { + s := newTestStores(t) + + item := &db.OutboundMessage{ + MessageID: "", + FromAddr: "a@test.local", + ToAddr: "b@fake.test", + RecipientDom: "fake.test", + RawData: "raw", + Status: db.OutboundStatusPending, + NextAttemptAt: time.Now(), + } + if err := s.Outbound.Create(item); err != nil { + t.Fatalf("create: %v", err) + } + + const n = 10 + var wins int64 + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ok, err := s.Outbound.Claim(item.ID) + if err != nil { + t.Errorf("claim: %v", err) + return + } + if ok { + atomic.AddInt64(&wins, 1) + } + }() + } + wg.Wait() + + if wins != 1 { + t.Fatalf("claims won = %d, want exactly 1", wins) + } + got, err := s.Outbound.GetByID(item.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.Status != db.OutboundStatusSending { + t.Fatalf("status = %s, want sending", got.Status) + } +} + +// TestOutboundClaimStatuses 验证只有 pending/deferred 可被抢占。 +func TestOutboundClaimStatuses(t *testing.T) { + s := newTestStores(t) + + cases := []struct { + status string + want bool + }{ + {db.OutboundStatusPending, true}, + {db.OutboundStatusDeferred, true}, + {db.OutboundStatusSending, false}, + {db.OutboundStatusSent, false}, + {db.OutboundStatusFailed, false}, + {db.OutboundStatusCanceled, false}, + } + for _, tc := range cases { + item := &db.OutboundMessage{ + MessageID: "", + FromAddr: "a@test.local", + ToAddr: "b@fake.test", + RecipientDom: "fake.test", + RawData: "raw", + Status: tc.status, + NextAttemptAt: time.Now(), + } + if err := s.Outbound.Create(item); err != nil { + t.Fatalf("create %s: %v", tc.status, err) + } + ok, err := s.Outbound.Claim(item.ID) + if err != nil { + t.Fatalf("claim %s: %v", tc.status, err) + } + if ok != tc.want { + t.Fatalf("claim %s = %v, want %v", tc.status, ok, tc.want) + } + if tc.want { + got, _ := s.Outbound.GetByID(item.ID) + if got.Status != db.OutboundStatusSending { + t.Fatalf("claimed %s item must become sending", tc.status) + } + } + } +}