feat(outbound): 外发投递多线程化（worker 池 + 每域并发上限）

- 1 个 dispatcher + N 个 worker 池（默认 4）并行投递，启动时立即
  扫描清空积压；workers=0/1 退化为串行（旧行为）
- 每收件域并发信号量（默认 2，配置 max_concurrent_per_domain），
  防对单个 MX 域并发过多被判定滥发；中继模式统一按 relay 限制
- store 新增 Claim 原子抢占（sending 状态防重复投递），并发下
  每封邮件恰好投递一次
- 新增配置 workers/batch_size（替代硬编码 50）；dispatcher 改为
  抢占后投递、worker 只负责发送，退避重试/退信/状态机不变
- 新增测试：真并发、串行回退、每域上限、临时/永久失败、错误隔离、
  Claim 并发原子性与状态可抢占性；-race 无警告

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                      # 单封邮件最大外部收件人数

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 秒扫描一次队列
+			PollInterval:           15, // 15 秒扫描一次队列
-			MaxAttempts:    12, // 最多尝试 12 次
+			MaxAttempts:            12, // 最多尝试 12 次
-			RetryBaseMin:   5,  // 5/10/20/40/... 分钟指数退避
+			RetryBaseMin:           5,  // 5/10/20/40/... 分钟指数退避
-			MaxRecipients:  50, // 单封最多 50 个外部收件人
+			MaxRecipients:          50, // 单封最多 50 个外部收件人
-			MaxPerMin:      30, // 每用户每分钟 30 封
+			MaxPerMin:              30, // 每用户每分钟 30 封
-			MaxPerDay:      500,
+			MaxPerDay:              500,
-			ConnectTimeout: 30,  // 连接远程 MX 超时 30 秒
+			ConnectTimeout:         30,  // 连接远程 MX 超时 30 秒
-			RelayPort:      587, // smarthost 默认提交端口
+			RelayPort:              587, // smarthost 默认提交端口
-			RelayStartTLS:  true,
+			RelayStartTLS:          true,
-			IPFamily:       "ipv4",
+			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
 	}

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"

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,
+// concurrent background delivery workers, exponential backoff retries, DKIM
-// per-user rate limits and failure bounces.
+// 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.
+// acquireDomain 获取收件域（或中继）的并发信号量，限制对同一目标域同时
-func (m *Manager) deliverOne(item *db.OutboundMessage) {
+// 打开的 SMTP 连接数。limit <= 0 表示不限制。
-	// Mark as sending to avoid concurrent workers double-delivering.
+func (m *Manager) acquireDomain(domain string) func() {
-	item.Status = db.OutboundStatusSending
+	limit := m.cfg.MaxConcurrentPerDomain
-	if err := m.stores.Outbound.Update(item); err != nil {
+	if limit <= 0 {
-		log.Printf("outbound: update item %d to sending failed: %v", item.ID, err)
+		return func() {}
-		return
 	}
 
-	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++

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:     "<seed@test>",
+			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)
+	}
+}

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

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:     "<t@test>",
+		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:     "<t@test>",
+			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)
+			}
+		}
+	}
+}