diff --git a/CHANGELOG.md b/CHANGELOG.md index d7e3648..84bea22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/internal/httpapi/qa_weekly_mail.go b/internal/httpapi/qa_weekly_mail.go index 6b91c04..961802a 100644 --- a/internal/httpapi/qa_weekly_mail.go +++ b/internal/httpapi/qa_weekly_mail.go @@ -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): diff --git a/internal/httpapi/qa_weekly_mail_pg_test.go b/internal/httpapi/qa_weekly_mail_pg_test.go new file mode 100644 index 0000000..2e2f210 --- /dev/null +++ b/internal/httpapi/qa_weekly_mail_pg_test.go @@ -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,'','',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 != "" || 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) + } + } +} diff --git a/internal/qawemail/resend_integration_test.go b/internal/qawemail/resend_integration_test.go index 2b71828..99d412a 100644 --- a/internal/qawemail/resend_integration_test.go +++ b/internal/qawemail/resend_integration_test.go @@ -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) { diff --git a/internal/qawemail/sends_readback_integration_test.go b/internal/qawemail/sends_readback_integration_test.go new file mode 100644 index 0000000..7cd53cc --- /dev/null +++ b/internal/qawemail/sends_readback_integration_test.go @@ -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) + } +} diff --git a/internal/qawemail/smtp_delivery_integration_test.go b/internal/qawemail/smtp_delivery_integration_test.go new file mode 100644 index 0000000..0125f92 --- /dev/null +++ b/internal/qawemail/smtp_delivery_integration_test.go @@ -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) + } +} diff --git a/internal/qawemail/smtp_mailer_test.go b/internal/qawemail/smtp_mailer_test.go new file mode 100644 index 0000000..7373ad1 --- /dev/null +++ b/internal/qawemail/smtp_mailer_test.go @@ -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: "

91.20%

", 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) + } + }) +} diff --git a/internal/qawemail/smtp_stub_test.go b/internal/qawemail/smtp_stub_test.go new file mode 100644 index 0000000..e7b1633 --- /dev/null +++ b/internal/qawemail/smtp_stub_test.go @@ -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 .") != 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} +} diff --git a/internal/qawemail/store.go b/internal/qawemail/store.go index dc0cc93..592898c 100644 --- a/internal/qawemail/store.go +++ b/internal/qawemail/store.go @@ -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 diff --git a/scripts/test-scope.json b/scripts/test-scope.json index 7ba91a5..7614399 100644 --- a/scripts/test-scope.json +++ b/scripts/test-scope.json @@ -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 不提供" } ] }