fix(qa-weekly-mail): 修复发送记录读回并区分未知投递
build / build (push) Successful in 2m18s

1) 发送记录读回(确定性回归):Sends() 的 SELECT 只有 6 列,Scan 却收 7 个字段(多出
message_id)。只要历史里存在任意一条发送记录,整个列表就返回 ErrUnavailable —— 发送
成功后页面刷新、以及“完成读真实发送记录”都会失败。改为查询带 message_id,并检查
rows.Err(),不把截断的列表当成完整历史。

2) 未知投递不再当成可重试失败:DATA 已提交但没收到最终 250(连接中断/超时)时,邮件可能
已被服务器接受;旧实现记 failed 并允许下一分钟重发,会重复投递。现在由 SMTPMailer 在
writer.Close() 失败处抛出窄类型 unknownDeliveryError,deliver 保留 sending 并记录原因,
自动/手动路径都不再重发(手动返回 ERRDeliveryUnknown,HTTP 409 说明原因)。连接建立前
失败、STARTTLS/AUTH/MAIL/RCPT/DATA 命令被拒、正文写入失败仍是确定失败,可如实 failed 并
重试一次。

3) SMTP 交互加有限期限:整体 90 秒(随请求 ctx 收紧),并用 context.AfterFunc 在取消时打断
阻塞读写,避免服务器接受 DATA 后不响应时挂死 worker 与发送行。

验证:真实隔离 PG 先 RED 后 GREEN(非空 sent+failed 历史读回成功并带真实 message_id;
HTTP GET /weekly-mail/sends 非空历史 200);本机 TLS SMTP 桩 + 真实 SMTPMailer 覆盖成功、
RCPT 550、DATA 后丢最终 250、DATA 阶段卡死(ctx 800ms 内返回)四类;真实 deliver 在未知
结果时留 sending 且不重发,在明确拒绝时 failed 并可重试一次成功。make verify 17 项全绿。
This commit is contained in:
姚章浩
2026-11-09 01:06:09 +08:00
parent 719876d039
commit 5e850e96f5
10 changed files with 690 additions and 11 deletions
+22
View File
@@ -1,5 +1,27 @@
# Changelog
## 开发快照 0.1.369(周报发送记录读回与未知投递收口)
### Fixed
- 修复周报“发送记录”读取:查询只取 6 列却按 7 个字段扫描(多出 message_id),只要该周或历史
里存在任意一条发送记录,列表接口就整体报“周报邮件暂不可用”,发送成功后刷新页面必然报错。
现在查询与字段一一对应,并在读取出错时如实失败(不把截断的历史当成完整列表)。
- 区分“发送失败”与“投递结果未知”:DATA 已提交、但没收到服务器最终 250(连接中断/超时)时,
邮件可能已经进入收件人邮箱。此前的实现把它记成 failed 并允许下一分钟自动重发,可能重复投递。
现在这类结果保留“发送中(结果未知)”,自动与手工路径都不再重发,并提示先核对已发出的邮件;
明确拒绝(认证失败、RCPT 550、连接建立前失败)仍记为失败并可重试一次。
- SMTP 交互增加有限超时(默认 90 秒,随请求 ctx 收紧并可被取消打断),避免服务器接受 DATA
后不再响应时把 worker 和发送记录挂死。
### Tests
- 真实隔离 PostgreSQL:非空发送历史(sent + failed)下单周/全量读回必须成功并带回真实
message_id(旧实现先 RED:ErrUnavailable);HTTP 发送记录接口在非空历史下 200。
- 本机 TLS SMTP 桩 + 真实 SMTPMailer:成功、RCPT 550 明确拒绝、DATA 后丢失最终 250、
DATA 阶段卡死(受 ctx 期限约束)四类结果分类正确;真实 deliver 在未知结果时留“发送中”
且不再重发,在明确拒绝时记失败并允许重试一次成功。
## 开发快照 0.1.368(周报发送窗口与正式补发修复)
### Fixed
+2
View File
@@ -39,6 +39,8 @@ func weeklyMailSendError(w http.ResponseWriter, err error) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "未配置邮件服务器、发件人或收件人,无法补发"})
case errors.Is(err, qawemail.ErrSendInFlight):
writeJSON(w, http.StatusConflict, map[string]string{"error": "该周该版本有一次发送结果未知(记录停在发送中),为避免重复投递未重发;请先核对已发出的邮件"})
case errors.Is(err, qawemail.ErrDeliveryUnknown):
writeJSON(w, http.StatusConflict, map[string]string{"error": "SMTP 已提交但未收到最终确认,投递结果未知;为避免重复已停止重发,请先核对已发出的邮件"})
case errors.Is(err, qawemail.ErrNoPassword):
writeJSON(w, http.StatusConflict, map[string]string{"error": "SMTP 授权码缺失或无法解密,无法补发"})
case errors.Is(err, qawemail.ErrInvalid):
@@ -0,0 +1,91 @@
package httpapi
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
postgresstore "knowledge-governance/internal/platform/postgres"
"knowledge-governance/internal/qawemail"
)
// 页面 reload / “完成读真实发送记录”走的就是这个 handler。用真实 PG 的非空历史证明
// 产品读回路径返回 200 且带 message_id,而不是 503——“历史行存在就整个列表不可用”
// 正是上一版 Sends() 的确定性回归。
func TestWeeklyMailSendsHandlerReadsRealHistoryWithMessageID(t *testing.T) {
dsn := os.Getenv("KG_WEEKLY_MAIL_TEST_DSN")
if dsn == "" {
t.Skip("显式一次性周报邮件数据库未提供(KG_WEEKLY_MAIL_TEST_DSN):跳过不等于通过")
}
if !strings.Contains(dsn, "/kg_weekly_mail_test_") || strings.Contains(dsn, "/governance?") {
t.Fatal("拒绝非一次性周报测试库")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
t.Fatal("测试库连接参数无效(详情省略)")
}
defer pool.Close()
if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
store := qawemail.NewStore(pool, "fixture-http-history", os.Getenv("KG_WEEKLY_MAIL_TEST_KEY"), nil)
for _, statement := range []string{
`DELETE FROM qa_weekly_email_sends WHERE tenant_key=$1`,
} {
if _, err := pool.Exec(ctx, statement, "fixture-http-history"); err != nil {
t.Fatal(err)
}
}
if _, err := pool.Exec(ctx, `INSERT INTO qa_weekly_email_sends(tenant_key,week,draft_revision,body_sha256,status,attempts,message_id,last_error,created_at,updated_at)
VALUES($1,'2026-W45',16,$2,'sent',1,'<real@example.com>','',now(),now()),
($1,'2026-W44',7,$2,'failed',2,'','smtp RCPT 被拒: 550',now(),now() - interval '1 hour')`,
"fixture-http-history", strings.Repeat("a", 64)); err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/qa-ops/master/weekly-mail/sends?week=2026-W45", nil)
request = request.WithContext(context.WithValue(request.Context(), sessionKey{}, sessionIdentity{Username: "tester", Role: roleAdmin}))
recorder := httptest.NewRecorder()
qaWeeklyMailSends(recorder, request, Config{QAWeeklyMail: store})
if recorder.Code != http.StatusOK {
t.Fatalf("非空历史必须 200,status=%d body=%s", recorder.Code, recorder.Body.String())
}
var payload struct {
Data struct {
Sends []qawemail.Send `json:"sends"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &payload); err != nil {
t.Fatal(err)
}
if len(payload.Data.Sends) != 1 || payload.Data.Sends[0].MessageID != "<real@example.com>" || payload.Data.Sends[0].Status != "sent" {
t.Fatalf("发送记录必须带回真实 message_id:%+v", payload.Data.Sends)
}
// 全量历史(不带 week)同样必须可读,且 failed 行不伪造 message_id。
all := httptest.NewRequest(http.MethodGet, "/api/v1/qa-ops/master/weekly-mail/sends", nil)
all = all.WithContext(context.WithValue(all.Context(), sessionKey{}, sessionIdentity{Username: "tester", Role: roleAdmin}))
allRecorder := httptest.NewRecorder()
qaWeeklyMailSends(allRecorder, all, Config{QAWeeklyMail: store})
if allRecorder.Code != http.StatusOK {
t.Fatalf("全量历史必须 200,status=%d body=%s", allRecorder.Code, allRecorder.Body.String())
}
if err := json.Unmarshal(allRecorder.Body.Bytes(), &payload); err != nil {
t.Fatal(err)
}
if len(payload.Data.Sends) != 2 {
t.Fatalf("应返回 2 条历史,got %+v", payload.Data.Sends)
}
for _, item := range payload.Data.Sends {
if item.Status == "failed" && item.MessageID != "" {
t.Fatalf("failed 行不得有 message_id:%+v", item)
}
}
}
+10 -4
View File
@@ -54,9 +54,16 @@ func (m *countingMailer) count() int {
return m.calls
}
// weeklyMailFixture applies the real schema to an explicitly disposable database. Refusing to
// run without KG_WEEKLY_MAIL_TEST_DSN keeps CI honest: a skip there is not a pass.
func weeklyMailFixture(t *testing.T) (*Store, *pgxpool.Pool, *countingMailer) {
t.Helper()
mailer := &countingMailer{}
store, pool := weeklyMailFixtureWith(t, mailer)
return store, pool, mailer
}
// weeklyMailFixtureWith applies the real schema to an explicitly disposable database. Refusing to
// run without KG_WEEKLY_MAIL_TEST_DSN keeps CI honest: a skip there is not a pass.
func weeklyMailFixtureWith(t *testing.T, mailer Mailer) (*Store, *pgxpool.Pool) {
t.Helper()
dsn := os.Getenv("KG_WEEKLY_MAIL_TEST_DSN")
if dsn == "" {
@@ -76,7 +83,6 @@ func weeklyMailFixture(t *testing.T) (*Store, *pgxpool.Pool, *countingMailer) {
if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
t.Fatal(err)
}
mailer := &countingMailer{}
store := NewStore(pool, "fixture-weekly-resend", os.Getenv("KG_WEEKLY_MAIL_TEST_KEY"), mailer)
for _, statement := range []string{
`DELETE FROM qa_weekly_email_sends WHERE tenant_key=$1`,
@@ -88,7 +94,7 @@ func weeklyMailFixture(t *testing.T) (*Store, *pgxpool.Pool, *countingMailer) {
t.Fatal(err)
}
}
return store, pool, mailer
return store, pool
}
func seedWeeklyDraft(t *testing.T, pool *pgxpool.Pool, tenant, week string, revision int64, markdown string) {
@@ -0,0 +1,54 @@
package qawemail
import (
"context"
"testing"
)
// 产品读回入口:页面 reload / GET /weekly-mail/sends 都走 store.Sends()。
// 旧实现 SELECT 只有 6 列却 Scan 7 个字段(多出 message_id),任何一条历史行都会让整个
// 列表返回 ErrUnavailable —— 发送成功后刷新页面必然报错。这里用真实 PG 的非空历史复现。
func TestSendsReadsBackMessageIDFromNonEmptyHistory(t *testing.T) {
store, pool, _ := weeklyMailFixture(t)
ctx := context.Background()
md := "# 知识库周报 2026-W45\n\n正确率:91.20%\n"
seedWeeklyDraft(t, pool, store.tenant, "2026-W45", 16, md)
confirmation, err := store.Confirm(ctx, "2026-W45", "fixture-admin")
if err != nil {
t.Fatal(err)
}
enabledWeeklySettings(t, store, 7, 23)
now := weekLocal(t, 2026, 11, 9, 0)
sent, err := store.SendManual(ctx, "2026-W45", 16, confirmation.BodySHA256, "fixture-admin", now)
if err != nil || sent.Status != "sent" || sent.MessageID == "" {
t.Fatalf("准备真实发送记录失败:%+v %v", sent, err)
}
// 单周读回(页面 reload 的真实入口)。
list, err := store.Sends(ctx, "2026-W45")
if err != nil {
t.Fatalf("非空历史下 Sends(week) 必须成功,got err=%v", err)
}
if len(list) != 1 || list[0].Status != "sent" || list[0].MessageID != sent.MessageID {
t.Fatalf("发送记录必须带回 message_id:%+v(期望 %q)", list, sent.MessageID)
}
// 全量列表(不带 week 的历史视图)。
all, err := store.Sends(ctx, "")
if err != nil {
t.Fatalf("非空历史下 Sends(all) 必须成功,got err=%v", err)
}
if len(all) != 1 || all[0].MessageID != sent.MessageID || all[0].Revision != 16 {
t.Fatalf("全量发送记录内容不符:%+v", all)
}
// 历史里已存在 failed 行(last_error 非空)也不能让读取失败。
if _, err := pool.Exec(ctx, `INSERT INTO qa_weekly_email_sends(tenant_key,week,draft_revision,body_sha256,status,attempts,last_error,created_at,updated_at)
VALUES($1,'2026-W44',3,$2,'failed',2,'smtp 登录失败: 535',now(),now())`, store.tenant, confirmation.BodySHA256); err != nil {
t.Fatal(err)
}
list, err = store.Sends(ctx, "2026-W44")
if err != nil || len(list) != 1 || list[0].LastError == "" || list[0].MessageID != "" {
t.Fatalf("failed 历史行必须可读且不假造 message_id:%+v %v", list, err)
}
}
@@ -0,0 +1,111 @@
package qawemail
import (
"context"
"crypto/tls"
"errors"
"strings"
"testing"
)
func stubSMTPSettings(t *testing.T, store *Store, port int) {
t.Helper()
if _, err := store.SaveSettings(context.Background(), "fixture-admin", SettingsInput{
BaseRevision: 0, Host: "127.0.0.1", Port: port, Password: "app-password",
From: "kb@example.com", Recipients: []string{"lead@example.com", "pm@example.com"},
Cc: []string{"qa@example.com"}, Weekday: 7, Hour: 23, Enabled: true,
}); err != nil {
t.Fatal(err)
}
}
// 真实 SMTPMailer + 真实 deliver + 真实 PG:DATA 提交后最终确认丢失 ⇒ 记录留 sending、
// 手动/自动都不得再投一次。这不是“随便造一个 error”,桩必须真的收到 DATA。
func TestRealSMTPUnknownAfterDataStaysSendingAndNeverRetries(t *testing.T) {
stub := newSMTPStub(t)
stub.setMode(smtpStubLostAck)
store, pool := weeklyMailFixtureWith(t, SMTPMailer{TLS: &tls.Config{InsecureSkipVerify: true}})
ctx := context.Background()
md := "# 知识库周报 2026-W45\n\n正确率:91.20%\n"
seedWeeklyDraft(t, pool, store.tenant, "2026-W45", 16, md)
confirmation, err := store.Confirm(ctx, "2026-W45", "fixture-admin")
if err != nil {
t.Fatal(err)
}
stubSMTPSettings(t, store, stub.port())
result, err := store.SendManual(ctx, "2026-W45", 16, confirmation.BodySHA256, "fixture-admin", weekLocal(t, 2026, 11, 9, 0))
if !errors.Is(err, ErrDeliveryUnknown) {
t.Fatalf("最终确认丢失必须 ErrDeliveryUnknown,got %+v %v", result, err)
}
if result.Status != "unknown" || result.MessageID != "" {
t.Fatalf("unknown 结果不得带 message_id:%+v", result)
}
if stub.deliveries() != 1 {
t.Fatalf("桩必须真的收到 1 次 DATA,got %d", stub.deliveries())
}
var status, messageID, lastError string
if err := pool.QueryRow(ctx, `SELECT status,message_id,last_error FROM qa_weekly_email_sends WHERE tenant_key=$1 AND week='2026-W45'`, store.tenant).Scan(&status, &messageID, &lastError); err != nil {
t.Fatal(err)
}
if status != "sending" || messageID != "" || !strings.Contains(lastError, "最终确认") {
t.Fatalf("未知结果必须留在 sending 并记录原因:status=%q message_id=%q last_error=%q", status, messageID, lastError)
}
stub.setMode(smtpStubAccept)
if again, err := store.SendDue(ctx, weekLocal(t, 2026, 11, 8, 23)); err != nil || again.State != "in_flight" {
t.Fatalf("未知结果后自动路径必须 in_flight,got %+v %v", again, err)
}
if _, err := store.SendManual(ctx, "2026-W45", 16, confirmation.BodySHA256, "fixture-admin", weekLocal(t, 2026, 11, 9, 9)); !errors.Is(err, ErrSendInFlight) {
t.Fatalf("未知结果后手动补发必须拒绝,got %v", err)
}
if stub.deliveries() != 1 {
t.Fatalf("未知结果不得重发:DATA 次数 %d", stub.deliveries())
}
}
// 明确拒绝(RCPT 550)是确定失败:记 failed、可重试一次;重试成功后写 sent + message_id。
func TestRealSMTPExplicitRejectionIsFailedAndRetryOnce(t *testing.T) {
stub := newSMTPStub(t)
stub.setMode(smtpStubRejectRcpt)
store, pool := weeklyMailFixtureWith(t, SMTPMailer{TLS: &tls.Config{InsecureSkipVerify: true}})
ctx := context.Background()
md := "# 知识库周报 2026-W45\n\n正确率:91.20%\n"
seedWeeklyDraft(t, pool, store.tenant, "2026-W45", 16, md)
confirmation, err := store.Confirm(ctx, "2026-W45", "fixture-admin")
if err != nil {
t.Fatal(err)
}
stubSMTPSettings(t, store, stub.port())
failed, err := store.SendManual(ctx, "2026-W45", 16, confirmation.BodySHA256, "fixture-admin", weekLocal(t, 2026, 11, 9, 0))
if err != nil || failed.Status != "failed" || failed.MessageID != "" {
t.Fatalf("明确拒绝必须如实 failed:%+v %v", failed, err)
}
if !strings.Contains(failed.Reason, "550") {
t.Fatalf("失败原因必须带真实 SMTP 回复:%q", failed.Reason)
}
if stub.rejections() != 1 || stub.deliveries() != 0 {
t.Fatalf("拒绝场景不应产生 DATA:rejections=%d datas=%d", stub.rejections(), stub.deliveries())
}
var status, lastError string
if err := pool.QueryRow(ctx, `SELECT status,last_error FROM qa_weekly_email_sends WHERE tenant_key=$1 AND week='2026-W45'`, store.tenant).Scan(&status, &lastError); err != nil {
t.Fatal(err)
}
if status != "failed" || !strings.Contains(lastError, "550") {
t.Fatalf("failed 记录不符:status=%q last_error=%q", status, lastError)
}
stub.setMode(smtpStubAccept)
retried, err := store.SendManual(ctx, "2026-W45", 16, confirmation.BodySHA256, "fixture-admin", weekLocal(t, 2026, 11, 9, 9))
if err != nil || retried.Status != "sent" || retried.MessageID == "" {
t.Fatalf("failed 记录应允许重试并成功:%+v %v", retried, err)
}
if stub.deliveries() != 1 {
t.Fatalf("重试应恰好产生 1 次 DATA,got %d", stub.deliveries())
}
sends, err := store.Sends(ctx, "2026-W45")
if err != nil || len(sends) != 1 || sends[0].MessageID != retried.MessageID || sends[0].Status != "sent" {
t.Fatalf("产品读回入口必须返回 sent + message_id:%+v %v", sends, err)
}
}
+84
View File
@@ -0,0 +1,84 @@
package qawemail
import (
"context"
"crypto/tls"
"testing"
"time"
)
// 真实 SMTPMailer 打真实 TCP/TLS 桩:证明“成功 / 明确拒绝 / DATA 后最终确认丢失 / DATA 阶段卡死”
// 四类结果被正确分类——只有后两类里“结果可能已被接受”的那一类才算 unknown。
func TestSMTPMailerClassifiesDeliveryOutcomes(t *testing.T) {
stub := newSMTPStub(t)
mailer := SMTPMailer{TLS: &tls.Config{InsecureSkipVerify: true}} // 仅测试桩:自签证书
cfg := SMTPConfig{Host: "127.0.0.1", Port: stub.port()}
message := Message{From: "kb@example.com", To: []string{"lead@example.com"}, Subject: "知识库周报 2026-W45", HTML: "<p>91.20%</p>", Text: "91.20%"}
t.Run("server_accepted", func(t *testing.T) {
stub.setMode(smtpStubAccept)
id, err := mailer.Send(context.Background(), cfg, message)
if err != nil || id == "" {
t.Fatalf("成功投递必须返回 message-id:%q %v", id, err)
}
if IsUnknownDelivery(err) {
t.Fatal("成功不能标记为 unknown")
}
if stub.deliveries() != 1 {
t.Fatalf("桩应收到 1 次 DATA,got %d", stub.deliveries())
}
})
t.Run("rcpt_explicitly_rejected", func(t *testing.T) {
stub.setMode(smtpStubRejectRcpt)
before := stub.deliveries()
_, err := mailer.Send(context.Background(), cfg, message)
if err == nil {
t.Fatal("RCPT 550 必须报错")
}
if IsUnknownDelivery(err) {
t.Fatalf("明确拒绝不是 unknown:%v", err)
}
if stub.deliveries() != before {
t.Fatal("被拒的收件人不应产生 DATA")
}
})
t.Run("final_ack_lost_after_data", func(t *testing.T) {
stub.setMode(smtpStubLostAck)
before := stub.deliveries()
_, err := mailer.Send(context.Background(), cfg, message)
if err == nil || !IsUnknownDelivery(err) {
t.Fatalf("DATA 已提交但最终 250 丢失必须 unknown,got %v (unknown=%v)", err, IsUnknownDelivery(err))
}
if stub.deliveries() != before+1 {
t.Fatal("桩必须真的收到了 DATA,否则这不是“提交后丢失确认”的场景")
}
})
t.Run("data_stage_stall_is_bounded_and_unknown", func(t *testing.T) {
stub.setMode(smtpStubStall)
ctx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond)
defer cancel()
start := time.Now()
_, err := mailer.Send(ctx, cfg, message)
elapsed := time.Since(start)
if err == nil || !IsUnknownDelivery(err) {
t.Fatalf("DATA 阶段超时必须 unknown,got %v (unknown=%v)", err, IsUnknownDelivery(err))
}
if elapsed > 10*time.Second {
t.Fatalf("SMTP 必须受 ctx 期限约束,实际耗时 %s", elapsed)
}
})
// 连接建立前的失败仍可安全重试:不是 unknown。
t.Run("connection_refused_is_retryable_failure", func(t *testing.T) {
closed := newSMTPStub(t)
port := closed.port()
closed.Close()
_, err := mailer.Send(context.Background(), SMTPConfig{Host: "127.0.0.1", Port: port}, message)
if err == nil || IsUnknownDelivery(err) {
t.Fatalf("连接失败必须如实 failed(非 unknown),got %v", err)
}
})
}
+224
View File
@@ -0,0 +1,224 @@
package qawemail
import (
"bufio"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"math/big"
"net"
"strings"
"sync"
"testing"
"time"
)
// smtpStub is a real TCP/TLS SMTP server for tests: it exercises the production SMTPMailer
// instead of mocking the Mailer interface, so the DATA-phase classification is proven end to end.
type smtpStub struct {
ln net.Listener
cert tls.Certificate
mu sync.Mutex
mode string
datas int
rejected int
closed chan struct{}
once sync.Once
}
const (
smtpStubAccept = "accept"
smtpStubRejectRcpt = "reject_rcpt"
smtpStubLostAck = "lost_ack"
smtpStubStall = "stall"
)
func newSMTPStub(t *testing.T) *smtpStub {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
stub := &smtpStub{ln: listener, cert: stubCertificate(t), mode: smtpStubAccept, closed: make(chan struct{})}
go stub.serve()
t.Cleanup(stub.Close)
return stub
}
func (s *smtpStub) Close() {
s.once.Do(func() {
close(s.closed)
_ = s.ln.Close()
})
}
func (s *smtpStub) port() int { return s.ln.Addr().(*net.TCPAddr).Port }
func (s *smtpStub) setMode(mode string) {
s.mu.Lock()
s.mode = mode
s.mu.Unlock()
}
func (s *smtpStub) currentMode() string {
s.mu.Lock()
defer s.mu.Unlock()
return s.mode
}
// deliveries counts completed DATA uploads, whether or not the final 250 was delivered.
func (s *smtpStub) deliveries() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.datas
}
func (s *smtpStub) rejections() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.rejected
}
func (s *smtpStub) serve() {
for {
conn, err := s.ln.Accept()
if err != nil {
return
}
go s.handle(conn)
}
}
func (s *smtpStub) handle(raw net.Conn) {
defer raw.Close()
conn := raw
reader := bufio.NewReader(conn)
writer := bufio.NewWriter(conn)
reply := func(line string) error {
if _, err := writer.WriteString(line + "\r\n"); err != nil {
return err
}
return writer.Flush()
}
if reply("220 stub ESMTP") != nil {
return
}
secured := false
for {
line, err := reader.ReadString('\n')
if err != nil {
return
}
upper := strings.ToUpper(strings.TrimSpace(line))
switch {
case strings.HasPrefix(upper, "EHLO"), strings.HasPrefix(upper, "HELO"):
lines := []string{"250-stub"}
if !secured {
lines = append(lines, "250-STARTTLS")
}
lines = append(lines, "250 OK")
for _, item := range lines {
if reply(item) != nil {
return
}
}
case upper == "STARTTLS":
if reply("220 ready to start TLS") != nil {
return
}
securedConn := tls.Server(conn, &tls.Config{Certificates: []tls.Certificate{s.cert}, MinVersion: tls.VersionTLS12})
if err := securedConn.Handshake(); err != nil {
return
}
conn = securedConn
reader = bufio.NewReader(conn)
writer = bufio.NewWriter(conn)
secured = true
case strings.HasPrefix(upper, "MAIL FROM"):
_ = reply("250 OK")
case strings.HasPrefix(upper, "RCPT TO"):
if s.currentMode() == smtpStubRejectRcpt {
s.mu.Lock()
s.rejected++
s.mu.Unlock()
_ = reply("550 5.1.1 mailbox unavailable")
continue
}
_ = reply("250 OK")
case upper == "DATA":
if reply("354 end data with <CR><LF>.<CR><LF>") != nil {
return
}
if err := readSMTPData(reader); err != nil {
return
}
s.mu.Lock()
s.datas++
s.mu.Unlock()
switch s.currentMode() {
case smtpStubLostAck:
// Drop the connection instead of the final 250: the client cannot know whether
// the server queued the message.
return
case smtpStubStall:
select {
case <-s.closed:
case <-time.After(30 * time.Second):
}
return
default:
if reply("250 2.0.0 Ok: queued as stub-1") != nil {
return
}
}
case upper == "QUIT":
_ = reply("221 Bye")
return
default:
_ = reply("250 OK")
}
}
}
func readSMTPData(reader *bufio.Reader) error {
for {
line, err := reader.ReadString('\n')
if err != nil {
return err
}
if strings.TrimRight(line, "\r\n") == "." {
return nil
}
}
}
// stubCertificate builds a throwaway self-signed cert for 127.0.0.1. Tests trust it explicitly;
// production never sees it.
func stubCertificate(t *testing.T) tls.Certificate {
t.Helper()
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
t.Fatal(err)
}
template := x509.Certificate{
SerialNumber: big.NewInt(1),
Subject: pkix.Name{CommonName: "kg-smtp-stub"},
NotBefore: time.Now().Add(-time.Hour),
NotAfter: time.Now().Add(time.Hour),
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
IsCA: true,
BasicConstraintsValid: true,
}
der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
if err != nil {
t.Fatal(err)
}
return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
}
+72 -7
View File
@@ -36,8 +36,25 @@ var (
ErrSlotPending = errors.New("weekly mail slot has not arrived for the running week")
ErrNotConfigured = errors.New("weekly mail transport is not configured")
ErrSendInFlight = errors.New("weekly mail send outcome is unknown")
// ErrDeliveryUnknown is the honest answer when the SMTP exchange ended without a final
// verdict: the server may already have queued the message, so retrying could duplicate it.
ErrDeliveryUnknown = errors.New("weekly mail delivery outcome is unknown")
)
// unknownDeliveryError marks a failure where the server may already have accepted the message.
// Only the caller that knows the SMTP phase can raise it, so a plain I/O error never silently
// becomes "unknown" (and never silently becomes a retryable failure either).
type unknownDeliveryError struct{ err error }
func (e *unknownDeliveryError) Error() string { return e.err.Error() }
func (e *unknownDeliveryError) Unwrap() error { return e.err }
// IsUnknownDelivery reports whether the send may have reached the recipients.
func IsUnknownDelivery(err error) bool {
var unknown *unknownDeliveryError
return errors.As(err, &unknown)
}
// Settings never carries the SMTP password; HasPassword only says whether one is stored.
type Settings struct {
Host string `json:"host"`
@@ -414,7 +431,7 @@ func (s *Store) Sends(ctx context.Context, week string) ([]Send, error) {
if !s.valid() {
return nil, ErrUnavailable
}
rows, err := s.pool.Query(ctx, `SELECT week,draft_revision,status,attempts,last_error,updated_at FROM qa_weekly_email_sends WHERE tenant_key=$1 AND ($2='' OR week=$2) ORDER BY updated_at DESC LIMIT 50`, s.tenant, week)
rows, err := s.pool.Query(ctx, `SELECT week,draft_revision,status,attempts,last_error,message_id,updated_at FROM qa_weekly_email_sends WHERE tenant_key=$1 AND ($2='' OR week=$2) ORDER BY updated_at DESC LIMIT 50`, s.tenant, week)
if err != nil {
return nil, ErrUnavailable
}
@@ -427,6 +444,10 @@ func (s *Store) Sends(ctx context.Context, week string) ([]Send, error) {
}
out = append(out, item)
}
// A read that stopped early is not a successful list: callers would silently show partial history.
if err := rows.Err(); err != nil {
return nil, ErrUnavailable
}
return out, nil
}
@@ -653,6 +674,12 @@ func (s *Store) deliver(ctx context.Context, settings Settings, week string, rev
if len(reason) > 400 {
reason = reason[:400]
}
if IsUnknownDelivery(sendErr) {
// The message may already be queued at the server. Leaving the row in 'sending' makes
// every later attempt (automatic or manual) refuse instead of blindly re-delivering.
_, _ = s.pool.Exec(ctx, `UPDATE qa_weekly_email_sends SET last_error=$2,updated_at=now() WHERE id=$1`, sendID, reason)
return delivery{state: "unknown", reason: "SMTP 已提交但未收到最终确认,投递结果未知;为避免重复已停止自动重发,请核对已发出的邮件:" + reason}, nil
}
_, _ = s.pool.Exec(ctx, `UPDATE qa_weekly_email_sends SET status='failed',last_error=$2,updated_at=now() WHERE id=$1`, sendID, reason)
return delivery{state: "failed", reason: "发送失败:" + reason}, nil
}
@@ -804,6 +831,8 @@ func (s *Store) SendManual(ctx context.Context, week string, revision int64, has
return out, ErrNoPassword
case "in_flight":
return out, ErrSendInFlight
case "unknown":
return out, ErrDeliveryUnknown
}
return out, nil
}
@@ -849,24 +878,56 @@ table{border-collapse:collapse}h1:first-child{margin-top:0}
}
// SMTPMailer speaks SMTP directly: implicit TLS on 465, STARTTLS elsewhere. Plaintext is refused.
type SMTPMailer struct{}
// TLS is nil in production (system roots); an in-process test stub injects its own trust so the
// real mailer — not a mock — can be exercised end to end.
type SMTPMailer struct{ TLS *tls.Config }
func (SMTPMailer) Send(ctx context.Context, cfg SMTPConfig, msg Message) (string, error) {
// smtpExchangeTimeout bounds the whole SMTP conversation. Without it a server that accepts DATA
// and then stops responding would hang the worker (and hold the send row) indefinitely.
const smtpExchangeTimeout = 90 * time.Second
func (m SMTPMailer) tlsConfig(host string) *tls.Config {
cfg := &tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}
if m.TLS != nil {
cfg = m.TLS.Clone()
if cfg.ServerName == "" {
cfg.ServerName = host
}
if cfg.MinVersion == 0 {
cfg.MinVersion = tls.VersionTLS12
}
}
return cfg
}
func (m SMTPMailer) Send(ctx context.Context, cfg SMTPConfig, msg Message) (string, error) {
if cfg.Host == "" || cfg.Port <= 0 || len(msg.To) == 0 || !validAddress(msg.From) {
return "", ErrInvalid
}
deadline := time.Now().Add(smtpExchangeTimeout)
if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
deadline = ctxDeadline
}
address := net.JoinHostPort(cfg.Host, fmt.Sprint(cfg.Port))
dialer := &net.Dialer{Timeout: 15 * time.Second}
dialer := &net.Dialer{Timeout: 15 * time.Second, Deadline: deadline}
var conn net.Conn
var err error
if cfg.Port == 465 {
conn, err = tls.DialWithDialer(dialer, "tcp", address, &tls.Config{ServerName: cfg.Host, MinVersion: tls.VersionTLS12})
conn, err = tls.DialWithDialer(dialer, "tcp", address, m.tlsConfig(cfg.Host))
} else {
conn, err = dialer.DialContext(ctx, "tcp", address)
}
if err != nil {
// The connection never opened: nothing was submitted, so this is a retryable failure.
return "", fmt.Errorf("smtp 连接失败: %w", err)
}
if err := conn.SetDeadline(deadline); err != nil {
conn.Close()
return "", fmt.Errorf("smtp 设置超时失败: %w", err)
}
// Cancellation must interrupt a blocked read/write, not just the dial.
stop := context.AfterFunc(ctx, func() { _ = conn.SetDeadline(time.Now()) })
defer stop()
client, err := smtp.NewClient(conn, cfg.Host)
if err != nil {
conn.Close()
@@ -877,7 +938,7 @@ func (SMTPMailer) Send(ctx context.Context, cfg SMTPConfig, msg Message) (string
if ok, _ := client.Extension("STARTTLS"); !ok {
return "", errors.New("smtp 服务器不支持 STARTTLS,已拒绝明文发送")
}
if err := client.StartTLS(&tls.Config{ServerName: cfg.Host, MinVersion: tls.VersionTLS12}); err != nil {
if err := client.StartTLS(m.tlsConfig(cfg.Host)); err != nil {
return "", fmt.Errorf("smtp STARTTLS 失败: %w", err)
}
}
@@ -901,10 +962,14 @@ func (SMTPMailer) Send(ctx context.Context, cfg SMTPConfig, msg Message) (string
messageID := fmt.Sprintf("<%s@%s>", hex.EncodeToString(randomBytes(12)), domainOf(msg.From))
boundary := hex.EncodeToString(randomBytes(10))
if _, err = writer.Write([]byte(buildMIME(msg, messageID, boundary))); err != nil {
// The body never finished uploading, so the server cannot have accepted it.
return "", fmt.Errorf("smtp 写入失败: %w", err)
}
if err = writer.Close(); err != nil {
return "", fmt.Errorf("smtp 提交失败: %w", err)
// Close() sends the terminating "." and waits for the final 250. Losing that exchange
// (dropped connection, timeout) means the message may already be queued: it must stay
// unknown rather than be retried as a plain failure.
return "", &unknownDeliveryError{err: fmt.Errorf("smtp 提交后未收到最终确认: %w", err)}
}
_ = client.Quit()
return messageID, nil
+20
View File
@@ -627,6 +627,26 @@
"package": "knowledge-governance/internal/httpapi",
"test": "TestStartupReconcilePreservesFailedCapturesForExactReconcile",
"reason": "需显式一次性隔离 kg_library_test_ PostgreSQL:真实 failed 误判保留、审计与精确核对纵向路径,CI 无库显式跳过;父已在真实隔离库执行 PASS,CI 绿不代表本用例运行"
},
{
"package": "knowledge-governance/internal/qawemail",
"test": "TestSendsReadsBackMessageIDFromNonEmptyHistory",
"reason": "需要一次性 PostgreSQL 与测试主密钥(KG_WEEKLY_MAIL_TEST_DSN/KG_WEEKLY_MAIL_TEST_KEY),CI 不提供;本批已在隔离库 real PG 上先 RED(历史行导致 ErrUnavailable)后 GREEN"
},
{
"package": "knowledge-governance/internal/qawemail",
"test": "TestRealSMTPUnknownAfterDataStaysSendingAndNeverRetries",
"reason": "需要一次性 PostgreSQL + 测试主密钥 + 本机 TLS SMTP 桩;证明真实 SMTPMailer 在 DATA 后丢最终 250 时留 sending 且不重发,CI 无此环境"
},
{
"package": "knowledge-governance/internal/qawemail",
"test": "TestRealSMTPExplicitRejectionIsFailedAndRetryOnce",
"reason": "需要一次性 PostgreSQL + 测试主密钥 + 本机 TLS SMTP 桩;证明明确 550 记 failed 且可重试一次,CI 无此环境"
},
{
"package": "knowledge-governance/internal/httpapi",
"test": "TestWeeklyMailSendsHandlerReadsRealHistoryWithMessageID",
"reason": "需要一次性 PostgreSQL + 测试主密钥;验证 GET /weekly-mail/sends 在非空历史下返回 200 与真实 message_id,CI 不提供"
}
]
}