kevin
b158b8f1f5
- 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 无警告
103 lines
2.3 KiB
Go
103 lines
2.3 KiB
Go
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)
|
|
}
|
|
}
|
|
}
|
|
}
|