fix(smtp): 修复下游投递失败后 SMTP DATA 仍返回成功，导致上游误判为已发送 (#356)

* fix(smtp): propagate downstream delivery failures

* test(smtp): support Go 1.26 textproto errors

server/listen/smtp_server/delivery_error.go

@@ -0,0 +1,94 @@
+package smtp_server
+
+import (
+	"errors"
+	"net"
+	"net/textproto"
+	"sort"
+	"strings"
+	"unicode/utf8"
+
+	smtp "github.com/emersion/go-smtp"
+)
+
+const maxDeliverySMTPMessageBytes = 128
+
+// downstreamDeliverySMTPError turns the synchronous downstream delivery result
+// into the SMTP response returned to the submitting client. Unknown failures are
+// temporary so a transient network problem does not permanently drop a message.
+func downstreamDeliverySMTPError(deliveryErr error, domainErrors map[string]error) error {
+	if deliveryErr == nil {
+		return nil
+	}
+
+	permanent := len(domainErrors) > 0
+	for _, domainErr := range domainErrors {
+		if domainErr == nil || !isPermanentDeliveryError(domainErr) {
+			permanent = false
+			break
+		}
+	}
+
+	if permanent {
+		return &smtp.SMTPError{
+			Code:         550,
+			EnhancedCode: smtp.EnhancedCode{5, 0, 0},
+			Message:      deliverySMTPMessage("Downstream delivery permanently failed", deliveryErr, domainErrors),
+		}
+	}
+
+	return &smtp.SMTPError{
+		Code:         451,
+		EnhancedCode: smtp.EnhancedCode{4, 4, 0},
+		Message:      deliverySMTPMessage("Downstream delivery temporarily failed", deliveryErr, domainErrors),
+	}
+}
+
+func deliverySMTPMessage(prefix string, deliveryErr error, domainErrors map[string]error) string {
+	details := make([]string, 0, len(domainErrors))
+	domains := make([]string, 0, len(domainErrors))
+	for domain := range domainErrors {
+		domains = append(domains, domain)
+	}
+	sort.Strings(domains)
+	for _, domain := range domains {
+		if domainErr := domainErrors[domain]; domainErr != nil {
+			details = append(details, domain+": "+domainErr.Error())
+		}
+	}
+	if len(details) == 0 && deliveryErr != nil {
+		details = append(details, deliveryErr.Error())
+	}
+
+	message := prefix
+	if len(details) > 0 {
+		// SMTP replies must stay on one bounded line, even when a remote server
+		// returns malformed text. Retain useful detail for the submitting client.
+		message += ": " + strings.Join(strings.Fields(strings.Join(details, "; ")), " ")
+	}
+	if len(message) <= maxDeliverySMTPMessageBytes {
+		return message
+	}
+
+	limit := maxDeliverySMTPMessageBytes - len("...")
+	for len(message) > limit {
+		_, size := utf8.DecodeLastRuneInString(message)
+		message = message[:len(message)-size]
+	}
+	return message + "..."
+}
+
+func isPermanentDeliveryError(err error) bool {
+	var protocolErr *textproto.Error
+	if errors.As(err, &protocolErr) {
+		return protocolErr.Code >= 500 && protocolErr.Code <= 599
+	}
+
+	var smtpErr *smtp.SMTPError
+	if errors.As(err, &smtpErr) {
+		return smtpErr.Code >= 500 && smtpErr.Code <= 599
+	}
+
+	var dnsErr *net.DNSError
+	return errors.As(err, &dnsErr) && dnsErr.IsNotFound
+}

server/listen/smtp_server/delivery_error_test.go

@@ -0,0 +1,136 @@
+package smtp_server
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"net/textproto"
+	"strings"
+	"testing"
+
+	smtp "github.com/emersion/go-smtp"
+)
+
+func TestDownstreamDeliverySMTPError(t *testing.T) {
+	tests := []struct {
+		name         string
+		deliveryErr  error
+		domainErrors map[string]error
+		wantCode     int
+		wantEnhanced smtp.EnhancedCode
+		wantMessages []string
+	}{
+		{name: "success"},
+		{
+			name:         "network failure is temporary",
+			deliveryErr:  errors.New("delivery failed"),
+			domainErrors: map[string]error{"example.com": &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connection refused")}},
+			wantCode:     451,
+			wantEnhanced: smtp.EnhancedCode{4, 4, 0},
+			wantMessages: []string{"connection refused"},
+		},
+		{
+			name:         "missing domain details is temporary",
+			deliveryErr:  errors.New("delivery failed"),
+			wantCode:     451,
+			wantEnhanced: smtp.EnhancedCode{4, 4, 0},
+			wantMessages: []string{"delivery failed"},
+		},
+		{
+			name:        "downstream 4xx is temporary",
+			deliveryErr: errors.New("delivery failed"),
+			domainErrors: map[string]error{
+				"example.com": &textproto.Error{Code: 421, Msg: "try again later"},
+			},
+			wantCode:     451,
+			wantEnhanced: smtp.EnhancedCode{4, 4, 0},
+			wantMessages: []string{"421", "try again later"},
+		},
+		{
+			name:        "downstream 5xx is permanent",
+			deliveryErr: errors.New("delivery failed"),
+			domainErrors: map[string]error{
+				"example.com": fmt.Errorf("remote rejected recipient: %w", &textproto.Error{Code: 550, Msg: "mailbox unavailable"}),
+			},
+			wantCode:     550,
+			wantEnhanced: smtp.EnhancedCode{5, 0, 0},
+			wantMessages: []string{"550", "mailbox unavailable"},
+		},
+		{
+			name:        "nxdomain is permanent",
+			deliveryErr: errors.New("delivery failed"),
+			domainErrors: map[string]error{
+				"missing.example": &net.OpError{Op: "dial", Net: "tcp", Err: &net.DNSError{Err: "no such host", Name: "missing.example", IsNotFound: true}},
+			},
+			wantCode:     550,
+			wantEnhanced: smtp.EnhancedCode{5, 0, 0},
+			wantMessages: []string{"no such host"},
+		},
+		{
+			name:        "mixed permanent and temporary failures are temporary",
+			deliveryErr: errors.New("delivery failed"),
+			domainErrors: map[string]error{
+				"rejected.example": &textproto.Error{Code: 550, Msg: "mailbox unavailable"},
+				"slow.example":     &net.OpError{Op: "dial", Net: "tcp", Err: &net.DNSError{Err: "timeout", Name: "slow.example", IsTimeout: true}},
+			},
+			wantCode:     451,
+			wantEnhanced: smtp.EnhancedCode{4, 4, 0},
+			wantMessages: []string{"550", "mailbox unavailable"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := downstreamDeliverySMTPError(tt.deliveryErr, tt.domainErrors)
+			if tt.wantCode == 0 {
+				if err != nil {
+					t.Fatalf("downstreamDeliverySMTPError() = %v, want nil", err)
+				}
+				return
+			}
+
+			var smtpErr *smtp.SMTPError
+			if !errors.As(err, &smtpErr) {
+				t.Fatalf("downstreamDeliverySMTPError() type = %T, want *smtp.SMTPError", err)
+			}
+			if smtpErr.Code != tt.wantCode {
+				t.Errorf("SMTP code = %d, want %d", smtpErr.Code, tt.wantCode)
+			}
+			if smtpErr.EnhancedCode != tt.wantEnhanced {
+				t.Errorf("enhanced code = %v, want %v", smtpErr.EnhancedCode, tt.wantEnhanced)
+			}
+			for _, part := range tt.wantMessages {
+				if !strings.Contains(smtpErr.Message, part) {
+					t.Errorf("SMTP message = %q, want it to contain %q", smtpErr.Message, part)
+				}
+			}
+			if strings.ContainsAny(smtpErr.Message, "\r\n") {
+				t.Errorf("SMTP message must be one line, got %q", smtpErr.Message)
+			}
+			if len(smtpErr.Message) > maxDeliverySMTPMessageBytes {
+				t.Errorf("SMTP message is too long: %d bytes", len(smtpErr.Message))
+			}
+		})
+	}
+}
+
+func TestDeliverySMTPMessageSanitizesAndBoundsDetails(t *testing.T) {
+	message := deliverySMTPMessage(
+		"Downstream delivery temporarily failed",
+		errors.New("fallback"),
+		map[string]error{
+			"b.example": errors.New("line one\r\nline two"),
+			"a.example": errors.New(strings.Repeat("x", 200)),
+		},
+	)
+
+	if strings.ContainsAny(message, "\r\n") {
+		t.Fatalf("SMTP message must be one line, got %q", message)
+	}
+	if len(message) > maxDeliverySMTPMessageBytes {
+		t.Fatalf("SMTP message is too long: %d bytes", len(message))
+	}
+	if !strings.Contains(message, "a.example") {
+		t.Fatalf("SMTP message is not deterministically sorted: %q", message)
+	}
+}

server/listen/smtp_server/read_content.go

@@ -99,7 +99,7 @@ func (s *Session) Data(r io.Reader) error {
 		}
 
 		errMsg := ""
-		err, sendErr := send.Send(ctx, email)
+		deliveryErr, sendErr := send.Send(ctx, email)
 
 		log.WithContext(ctx).Debugf("插件执行--SendAfter")
 
@@ -115,8 +115,8 @@ func (s *Session) Data(r io.Reader) error {
 		as3.Wait()
 		log.WithContext(ctx).Debugf("插件执行--SendAfter")
 
-		if err != nil {
-			errMsg = err.Error()
+		if deliveryErr != nil {
+			errMsg = deliveryErr.Error()
 			_, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =2 ,error=? where id = ? "), errMsg, email.MessageId)
 			if err != nil {
 				log.WithContext(ctx).Errorf("sql Error :%+v", err)
@@ -126,6 +126,8 @@ func (s *Session) Data(r io.Reader) error {
 				log.WithContext(ctx).Errorf("sql Error :%+v", err)
 			}
 
+			return downstreamDeliverySMTPError(deliveryErr, sendErr)
+
 		} else {
 			_, err := db.Instance.Exec(db.WithContext(ctx, "update email set status =1  where id = ? "), email.MessageId)
 			if err != nil {

server/utils/send/send.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"errors"
+	"fmt"
 	"github.com/Jinnrry/pmail/config"
 	"github.com/Jinnrry/pmail/dto/parsemail"
 	"github.com/Jinnrry/pmail/models"
@@ -14,6 +15,7 @@ import (
 	"github.com/Jinnrry/pmail/utils/smtp"
 	log "github.com/sirupsen/logrus"
 	"net"
+	"net/textproto"
 	"strings"
 	"sync"
 )
@@ -23,6 +25,19 @@ type mxDomain struct {
 	mxHost string
 }
 
+type temporaryMXFallbackError struct {
+	lookupErr   error
+	fallbackErr error
+}
+
+func (e *temporaryMXFallbackError) Error() string {
+	return fmt.Sprintf("temporary MX lookup failed (%v); fallback delivery failed: %v", e.lookupErr, e.fallbackErr)
+}
+
+func (e *temporaryMXFallbackError) Unwrap() error {
+	return e.lookupErr
+}
+
 // Forward 转发邮件
 func Forward(ctx *context.Context, e *parsemail.Email, forwardAddress string, user *models.User) error {
 
@@ -70,6 +85,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 
 	// 按域名整理
 	toByDomain := map[mxDomain][]*parsemail.User{}
+	mxLookupErrors := map[mxDomain]error{}
 	for _, s := range to {
 		args := strings.Split(s.EmailAddress, "@")
 		if len(args) == 2 {
@@ -82,13 +98,13 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 				toByDomain[address] = append(toByDomain[address], s)
 			} else {
 				//查询dns mx记录
-				mxInfo, err := net.LookupMX(args[1])
+				mxInfo, lookupErr := net.LookupMX(args[1])
 				address := mxDomain{
 					domain: "smtp." + args[1],
 					mxHost: "smtp." + args[1],
 				}
-				if err != nil {
-					log.WithContext(ctx).Errorf(s.EmailAddress, "域名mx记录查询失败，检查邮箱是否存在！")
+				if lookupErr != nil {
+					log.WithContext(ctx).Errorf("%s 域名mx记录查询失败，检查邮箱是否存在！", s.EmailAddress)
 				}
 				if len(mxInfo) > 0 {
 					address = mxDomain{
@@ -96,6 +112,9 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 						mxHost: mxInfo[0].Host,
 					}
 				}
+				if lookupErr != nil {
+					mxLookupErrors[address] = lookupErr
+				}
 				toByDomain[address] = append(toByDomain[address], s)
 			}
 		} else {
@@ -105,6 +124,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 	}
 
 	var errEmailAddress []string
+	var errEmailAddressMu sync.Mutex
 
 	errMap := sync.Map{}
 
@@ -112,12 +132,25 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 	for domain, tos := range toByDomain {
 		domain := domain
 		tos := tos
+		mxLookupErr := mxLookupErrors[domain]
 		as.WaitProcess(func(p any) {
+			recordFailure := func(err error) {
+				err = deliveryFailureCause(mxLookupErr, err)
+				log.WithContext(ctx).Errorf("%v 邮件投递失败%+v", tos, err)
+
+				errEmailAddressMu.Lock()
+				for _, user := range tos {
+					errEmailAddress = append(errEmailAddress, user.EmailAddress)
+				}
+				errEmailAddressMu.Unlock()
+
+				errMap.Store(domain.domain, err)
+			}
 
 			if domain.domain == "localhost" {
 				err := smtp.SendMailUnsafe("", domain.mxHost+":25", nil, from, fromDomain, buildAddress(tos), data)
 				if err != nil {
-					log.WithContext(ctx).Errorf("send error %s", err.Error())
+					recordFailure(err)
 				}
 				return
 			}
@@ -143,6 +176,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 			if err == nil {
 				return
 			}
+			if isPermanentSMTPResponse(err) {
+				recordFailure(err)
+				return
+			}
 			log.WithContext(ctx).Infof("SMTP STARTTLS on 25 Send Error. %s", err.Error())
 
 			// 再试用587投递
@@ -150,6 +187,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 			if err == nil {
 				return
 			}
+			if isPermanentSMTPResponse(err) {
+				recordFailure(err)
+				return
+			}
 			log.WithContext(ctx).Infof("SMTPS on 587 Send Error. %s", err.Error())
 
 			// 再次尝试465投递
@@ -157,6 +198,10 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 			if err == nil {
 				return
 			}
+			if isPermanentSMTPResponse(err) {
+				recordFailure(err)
+				return
+			}
 			log.WithContext(ctx).Infof("SMTPS on 465 Send Error. %s", err.Error())
 
 			// 最后尝试非安全方式投递
@@ -166,13 +211,7 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 				return
 			}
 
-			if err != nil {
-				log.WithContext(ctx).Errorf("%v 邮件投递失败%+v", tos, err)
-				for _, user := range tos {
-					errEmailAddress = append(errEmailAddress, user.EmailAddress)
-				}
-			}
-			errMap.Store(domain.domain, err)
+			recordFailure(err)
 		}, nil)
 	}
 	as.Wait()
@@ -194,6 +233,32 @@ func doSend(ctx *context.Context, fromDomain string, data []byte, to []*parsemai
 	return nil, orgMap
 }
 
+func isPermanentSMTPResponse(err error) bool {
+	var protocolErr *textproto.Error
+	return errors.As(err, &protocolErr) && protocolErr.Code >= 500 && protocolErr.Code <= 599
+}
+
+func deliveryFailureCause(mxLookupErr, fallbackErr error) error {
+	if fallbackErr == nil || isPermanentSMTPResponse(fallbackErr) {
+		return fallbackErr
+	}
+
+	var lookupDNSErr *net.DNSError
+	if !errors.As(mxLookupErr, &lookupDNSErr) || (!lookupDNSErr.IsTimeout && !lookupDNSErr.IsTemporary) {
+		return fallbackErr
+	}
+
+	var networkErr net.Error
+	if !errors.As(fallbackErr, &networkErr) {
+		return fallbackErr
+	}
+
+	return &temporaryMXFallbackError{
+		lookupErr:   mxLookupErr,
+		fallbackErr: fallbackErr,
+	}
+}
+
 func buildAddress(u []*parsemail.User) []string {
 	var ret []string
 

server/utils/send/send_test.go

@@ -0,0 +1,80 @@
+package send
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"net/textproto"
+	"testing"
+)
+
+func TestIsPermanentSMTPResponse(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{name: "nil"},
+		{name: "network error", err: errors.New("connection refused")},
+		{name: "temporary SMTP response", err: &textproto.Error{Code: 451, Msg: "try later"}},
+		{name: "wrapped temporary SMTP response", err: fmt.Errorf("delivery: %w", &textproto.Error{Code: 421, Msg: "service unavailable"})},
+		{name: "permanent SMTP response", err: &textproto.Error{Code: 550, Msg: "mailbox unavailable"}, want: true},
+		{name: "wrapped permanent SMTP response", err: fmt.Errorf("delivery: %w", &textproto.Error{Code: 554, Msg: "transaction failed"}), want: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isPermanentSMTPResponse(tt.err); got != tt.want {
+				t.Fatalf("isPermanentSMTPResponse() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDeliveryFailureCausePreservesTemporaryMXError(t *testing.T) {
+	tests := []struct {
+		name      string
+		lookupErr *net.DNSError
+	}{
+		{
+			name:      "SERVFAIL before fallback NXDOMAIN",
+			lookupErr: &net.DNSError{Err: "server misbehaving", Name: "example.com", IsTemporary: true},
+		},
+		{
+			name:      "timeout before fallback NXDOMAIN",
+			lookupErr: &net.DNSError{Err: "i/o timeout", Name: "example.com", IsTimeout: true},
+		},
+	}
+
+	fallbackErr := &net.OpError{
+		Op:  "dial",
+		Net: "tcp",
+		Err: &net.DNSError{Err: "no such host", Name: "smtp.example.com", IsNotFound: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := deliveryFailureCause(tt.lookupErr, fallbackErr)
+
+			var dnsErr *net.DNSError
+			if !errors.As(err, &dnsErr) {
+				t.Fatalf("deliveryFailureCause() = %T, want wrapped DNS error", err)
+			}
+			if dnsErr.IsNotFound {
+				t.Fatalf("deliveryFailureCause() selected fallback NXDOMAIN: %v", dnsErr)
+			}
+			if !dnsErr.IsTemporary && !dnsErr.IsTimeout {
+				t.Fatalf("deliveryFailureCause() selected non-temporary DNS error: %v", dnsErr)
+			}
+		})
+	}
+}
+
+func TestDeliveryFailureCauseKeepsExplicitSMTPRejection(t *testing.T) {
+	lookupErr := &net.DNSError{Err: "server misbehaving", Name: "example.com", IsTemporary: true}
+	rejection := &textproto.Error{Code: 550, Msg: "mailbox unavailable"}
+
+	if got := deliveryFailureCause(lookupErr, rejection); got != rejection {
+		t.Fatalf("deliveryFailureCause() = %v, want explicit SMTP rejection %v", got, rejection)
+	}
+}

server/utils/smtp/smtp.go

@@ -36,6 +36,8 @@ import (
 var NoSupportSTARTTLSError = errors.New("smtp: server doesn't support STARTTLS")
 var EOFError = errors.New("EOF")
 
+const outboundSMTPQuitWriteTimeout = 250 * time.Millisecond
+
 // A Client represents a client connection to an SMTP server.
 type Client struct {
 	// Text is the textproto.Conn used by the Client. It is exported to allow for
@@ -341,6 +343,26 @@ func (c *Client) Data() (io.WriteCloser, error) {
 	return &dataCloser{c, c.Text.DotWriter()}, nil
 }
 
+func finishDelivery(c *Client, w io.WriteCloser, msg []byte) error {
+	if _, err := w.Write(msg); err != nil {
+		return err
+	}
+	if err := w.Close(); err != nil {
+		return err
+	}
+
+	// The final 250 response to DATA transfers responsibility for the message.
+	// Send QUIT with a bounded write, but do not wait for 221: a failed or
+	// missing reply after acceptance must not block or trigger redelivery.
+	if err := c.conn.SetWriteDeadline(time.Now().Add(outboundSMTPQuitWriteTimeout)); err != nil {
+		log.Debugf("Could not bound SMTP QUIT write after message acceptance: %v", err)
+	}
+	if err := c.Text.PrintfLine("QUIT"); err != nil {
+		log.Debugf("SMTP QUIT failed after message acceptance: %v", err)
+	}
+	return nil
+}
+
 func SendMailWithTls(domain string, addr string, a smtp.Auth, from string, fromDomain string, to []string, msg []byte) error {
 	if err := validateLine(from); err != nil {
 		return err
@@ -378,15 +400,7 @@ func SendMailWithTls(domain string, addr string, a smtp.Auth, from string, fromD
 	if err != nil {
 		return err
 	}
-	_, err = w.Write(msg)
-	if err != nil {
-		return err
-	}
-	err = w.Close()
-	if err != nil {
-		return err
-	}
-	return c.Quit()
+	return finishDelivery(c, w, msg)
 }
 
 // SendMail connects to the server at addr, switches to TLS if
@@ -464,15 +478,7 @@ func SendMail(domain string, addr string, a smtp.Auth, from string, fromDomain s
 	if err != nil {
 		return err
 	}
-	_, err = w.Write(msg)
-	if err != nil {
-		return err
-	}
-	err = w.Close()
-	if err != nil {
-		return err
-	}
-	return c.Quit()
+	return finishDelivery(c, w, msg)
 }
 
 // SendMailUnsafe 无TLS加密的邮件发送方式
@@ -514,15 +520,7 @@ func SendMailUnsafe(domain string, addr string, a smtp.Auth, from string, fromDo
 	if err != nil {
 		return err
 	}
-	_, err = w.Write(msg)
-	if err != nil {
-		return err
-	}
-	err = w.Close()
-	if err != nil {
-		return err
-	}
-	return c.Quit()
+	return finishDelivery(c, w, msg)
 }
 
 // Extension reports whether an extension is support by the server.

server/utils/smtp/smtp_test.go

@@ -0,0 +1,182 @@
+package smtp
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/textproto"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSendMailUnsafeUsesFinalDataResponseAsDeliveryResult(t *testing.T) {
+	tests := []struct {
+		name                 string
+		dataResponse         string
+		holdWithoutQuitReply bool
+		wantCode             int
+		maxDuration          time.Duration
+	}{
+		{
+			name:                 "accepted message returns while server withholds QUIT reply",
+			dataResponse:         "250 2.0.0 queued",
+			holdWithoutQuitReply: true,
+			maxDuration:          time.Second,
+		},
+		{
+			name:         "DATA rejection remains a delivery failure",
+			dataResponse: "550 5.0.0 rejected",
+			wantCode:     550,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			addr, serverDone := startSMTPTransactionServer(t, tt.dataResponse, tt.holdWithoutQuitReply)
+
+			start := time.Now()
+			err := SendMailUnsafe(
+				"",
+				addr,
+				nil,
+				"sender@example.com",
+				"example.com",
+				[]string{"recipient@example.net"},
+				[]byte("From: sender@example.com\r\nTo: recipient@example.net\r\nSubject: test\r\n\r\nbody\r\n"),
+			)
+			duration := time.Since(start)
+			if tt.maxDuration > 0 && duration > tt.maxDuration {
+				t.Fatalf("SendMailUnsafe() took %v, want at most %v", duration, tt.maxDuration)
+			}
+
+			if tt.wantCode == 0 {
+				if err != nil {
+					t.Fatalf("SendMailUnsafe() = %v, want nil", err)
+				}
+			} else {
+				var protocolErr *textproto.Error
+				if !errors.As(err, &protocolErr) {
+					t.Fatalf("SendMailUnsafe() error type = %T, want *textproto.Error", err)
+				}
+				if protocolErr.Code != tt.wantCode {
+					t.Fatalf("SMTP code = %d, want %d", protocolErr.Code, tt.wantCode)
+				}
+			}
+
+			select {
+			case serverErr := <-serverDone:
+				if serverErr != nil {
+					t.Fatal(serverErr)
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatal("fake SMTP server did not finish")
+			}
+		})
+	}
+}
+
+func startSMTPTransactionServer(t *testing.T, dataResponse string, holdWithoutQuitReply bool) (string, <-chan error) {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		defer listener.Close()
+
+		conn, err := listener.Accept()
+		if err != nil {
+			done <- err
+			return
+		}
+		defer conn.Close()
+		_ = conn.SetDeadline(time.Now().Add(2 * time.Second))
+
+		text := textproto.NewConn(conn)
+		defer text.Close()
+
+		if err := text.PrintfLine("220 test ESMTP ready"); err != nil {
+			done <- err
+			return
+		}
+		if err := expectSMTPCommand(text, "EHLO "); err != nil {
+			done <- err
+			return
+		}
+		if err := text.PrintfLine("250 test"); err != nil {
+			done <- err
+			return
+		}
+		if err := expectSMTPCommand(text, "MAIL FROM:"); err != nil {
+			done <- err
+			return
+		}
+		if err := text.PrintfLine("250 2.1.0 sender accepted"); err != nil {
+			done <- err
+			return
+		}
+		if err := expectSMTPCommand(text, "RCPT TO:"); err != nil {
+			done <- err
+			return
+		}
+		if err := text.PrintfLine("250 2.1.5 recipient accepted"); err != nil {
+			done <- err
+			return
+		}
+		if err := expectSMTPCommand(text, "DATA"); err != nil {
+			done <- err
+			return
+		}
+		if err := text.PrintfLine("354 end data with <CR><LF>.<CR><LF>"); err != nil {
+			done <- err
+			return
+		}
+		body, err := io.ReadAll(text.DotReader())
+		if err != nil {
+			done <- err
+			return
+		}
+		if !strings.Contains(string(body), "Subject: test") {
+			done <- fmt.Errorf("message body was not received: %q", body)
+			return
+		}
+		if err := text.PrintfLine("%s", dataResponse); err != nil {
+			done <- err
+			return
+		}
+
+		if holdWithoutQuitReply {
+			if err := expectSMTPCommand(text, "QUIT"); err != nil {
+				done <- err
+				return
+			}
+
+			// Keep the server side open without sending 221. The client must close
+			// the accepted transaction without waiting for a reply.
+			buf := make([]byte, 1)
+			if _, err := conn.Read(buf); err != io.EOF {
+				done <- fmt.Errorf("waiting for client close after QUIT: %w", err)
+				return
+			}
+		}
+		done <- nil
+	}()
+
+	return listener.Addr().String(), done
+}
+
+func expectSMTPCommand(conn *textproto.Conn, prefix string) error {
+	line, err := conn.ReadLine()
+	if err != nil {
+		return err
+	}
+	if !strings.HasPrefix(line, prefix) {
+		return fmt.Errorf("SMTP command = %q, want prefix %q", line, prefix)
+	}
+	return nil
+}