From dfbc47d1724444d465debd8e542caf5ba3594388 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 10:26:48 +0100 Subject: [PATCH 1/9] framework/module: Add optional StreamingDelivery interface Delivery.Body takes a buffer.Buffer, which can be reopened and has a known length, so a message source has to buffer the whole body before passing it on. A target that only copies the body somewhere once does not need that. StreamingDelivery lets such a target receive the body as an io.Reader while the source is still receiving it. It is optional, so existing targets and the buffer.Buffer contract are unchanged. CanStreamBody is separate from BodyStream because whether streaming is possible can depend on the recipients added to the delivery. --- framework/module/streaming_delivery.go | 44 ++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 framework/module/streaming_delivery.go diff --git a/framework/module/streaming_delivery.go b/framework/module/streaming_delivery.go new file mode 100644 index 00000000..2ca0ffd5 --- /dev/null +++ b/framework/module/streaming_delivery.go @@ -0,0 +1,44 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package module + +import ( + "context" + "io" + + "github.com/emersion/go-message/textproto" +) + +// StreamingDelivery is an optional interface that may be implemented by the +// object returned by DeliveryTarget.StartDelivery to receive the message body +// while it is still being read from the source, without buffering it first. +type StreamingDelivery interface { + // CanStreamBody reports whether BodyStream can be used instead of Body for + // the current message. It is called after all AddRcpt calls. + CanStreamBody() bool + + // BodyStream is similar to Body of the regular Delivery interface except + // that the body is read exactly once, while the message source is still + // receiving it. + // + // If reading body fails, BodyStream must return an error and must make + // sure the partially read message is not delivered. The message source + // calls Abort after a failed BodyStream. + BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error +} From 81ca27019c7cc53cfcde2c2188ccf8155b5ff524 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 10:26:48 +0100 Subject: [PATCH 2/9] target/smtp: Implement StreamingDelivery The body is copied into the DATA command as it is read. If reading it fails midway, the connection is dropped instead of closed with QUIT: net/textproto terminates an open DATA before writing the next command, so QUIT would make the downstream server accept the partial message. testutils.Target gets an option to implement StreamingDelivery, and DoTestDeliveryStream mirrors DoTestDelivery for it. --- internal/target/smtp/smtp_downstream.go | 41 ++++++++++ internal/target/smtp/smtp_downstream_test.go | 64 +++++++++++++++ internal/testutils/target.go | 82 ++++++++++++++++++++ 3 files changed, 187 insertions(+) diff --git a/internal/target/smtp/smtp_downstream.go b/internal/target/smtp/smtp_downstream.go index 823b1032..48926aa5 100644 --- a/internal/target/smtp/smtp_downstream.go +++ b/internal/target/smtp/smtp_downstream.go @@ -30,6 +30,7 @@ import ( "context" "crypto/tls" "fmt" + "io" "net" "runtime/trace" "time" @@ -176,6 +177,8 @@ type delivery struct { rcpts []string conn *smtpconn.C + + connDropped bool } // lmtpDelivery implements module.PartialDelivery @@ -301,6 +304,38 @@ func (d *delivery) Body(ctx context.Context, header textproto.Header, body buffe return d.u.moduleError(d.conn.Data(ctx, header, r)) } +func (d *delivery) CanStreamBody() bool { + return true +} + +type errorRecordingReader struct { + r io.Reader + err error +} + +func (r *errorRecordingReader) Read(b []byte) (int, error) { + n, err := r.r.Read(b) + if err != nil && err != io.EOF { + r.err = err + } + return n, err +} + +func (d *delivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + r := &errorRecordingReader{r: body} + err := d.conn.Data(ctx, header, r) + if r.err != nil { + // Sending QUIT would first terminate the DATA command and make the + // server accept the partial message. Drop the connection instead. + d.connDropped = true + if err := d.conn.DirectClose(); err != nil { + d.log.Error("failed to close SMTP connection", err) + } + return r.err + } + return d.u.moduleError(err) +} + func (d *lmtpDelivery) BodyNonAtomic(ctx context.Context, sc module.StatusCollector, header textproto.Header, body buffer.Buffer) { r, err := body.Open() if err != nil { @@ -339,10 +374,16 @@ func (d *lmtpDelivery) BodyNonAtomic(ctx context.Context, sc module.StatusCollec } func (d *delivery) Abort(ctx context.Context) error { + if d.connDropped { + return nil + } return d.conn.Close() } func (d *delivery) Commit(ctx context.Context) error { + if d.connDropped { + return nil + } return d.conn.Close() } diff --git a/internal/target/smtp/smtp_downstream_test.go b/internal/target/smtp/smtp_downstream_test.go index e292f5e3..ad2505a6 100644 --- a/internal/target/smtp/smtp_downstream_test.go +++ b/internal/target/smtp/smtp_downstream_test.go @@ -19,16 +19,22 @@ along with this program. If not, see . package smtp_downstream import ( + "context" "errors" "flag" + "io" "math/rand" "os" "strconv" + "strings" "testing" + "testing/iotest" + "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" "github.com/foxcpp/maddy/framework/config" "github.com/foxcpp/maddy/framework/exterrors" + "github.com/foxcpp/maddy/framework/module" "github.com/foxcpp/maddy/internal/testutils" "github.com/stretchr/testify/require" ) @@ -276,6 +282,64 @@ func TestDownstreamDelivery_StartTLS_NoFallback(t *testing.T) { } } +func TestDownstreamDelivery_BodyStream(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+testPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + + mod := &Downstream{ + hostname: "mx.example.invalid", + endpoints: []config.Endpoint{ + { + Scheme: "tcp", + Host: "127.0.0.1", + Port: testPort, + }, + }, + log: testutils.Logger(t, "target.smtp"), + } + + testutils.DoTestDeliveryStream(t, mod, "test@example.invalid", []string{"rcpt@example.invalid"}) + be.CheckMsg(t, 0, "test@example.invalid", []string{"rcpt@example.invalid"}) +} + +func TestDownstreamDelivery_BodyStream_ReadErr(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+testPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + + mod := &Downstream{ + hostname: "mx.example.invalid", + endpoints: []config.Endpoint{ + { + Scheme: "tcp", + Host: "127.0.0.1", + Port: testPort, + }, + }, + log: testutils.Logger(t, "target.smtp"), + } + + ctx := context.Background() + delivery, err := mod.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.invalid") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "rcpt@example.invalid", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + + require.NoError(t, delivery.Abort(ctx)) + + testutils.WaitForConnsClose(t, srv) + require.Empty(t, be.Messages) +} + func TestMain(m *testing.M) { remoteSmtpPort := flag.String("test.smtpport", "random", "(maddy) SMTP port to use for connections in tests") flag.Parse() diff --git a/internal/testutils/target.go b/internal/testutils/target.go index 221834ba..f8b04990 100644 --- a/internal/testutils/target.go +++ b/internal/testutils/target.go @@ -26,6 +26,7 @@ import ( "io" "reflect" "sort" + "strings" "testing" "github.com/emersion/go-message/textproto" @@ -42,6 +43,7 @@ type Msg struct { RcptTo []string Body []byte Header textproto.Header + Streamed bool } type Target struct { @@ -55,6 +57,9 @@ type Target struct { AbortErr error CommitErr error + // Streaming makes deliveries implement module.StreamingDelivery. + Streaming bool + InstName string } @@ -86,6 +91,10 @@ type testTargetDeliveryPartial struct { testTargetDelivery } +type testTargetDeliveryStreaming struct { + testTargetDelivery +} + func (dt *Target) StartDelivery(ctx context.Context, msgMeta *module.MsgMetadata, mailFrom string) (module.Delivery, error) { if dt.PartialBodyErr != nil { return &testTargetDeliveryPartial{ @@ -95,6 +104,14 @@ func (dt *Target) StartDelivery(ctx context.Context, msgMeta *module.MsgMetadata }, }, dt.StartErr } + if dt.Streaming { + return &testTargetDeliveryStreaming{ + testTargetDelivery: testTargetDelivery{ + tgt: dt, + msg: Msg{MsgMeta: msgMeta, MailFrom: mailFrom}, + }, + }, dt.StartErr + } return &testTargetDelivery{ tgt: dt, msg: Msg{MsgMeta: msgMeta, MailFrom: mailFrom}, @@ -173,6 +190,23 @@ func (dtd *testTargetDelivery) Body(ctx context.Context, header textproto.Header return err } +func (dtd *testTargetDeliveryStreaming) CanStreamBody() bool { + return true +} + +func (dtd *testTargetDeliveryStreaming) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + if dtd.tgt.BodyErr != nil { + return dtd.tgt.BodyErr + } + + dtd.msg.Header = header + dtd.msg.Streamed = true + + var err error + dtd.msg.Body, err = io.ReadAll(body) + return err +} + func (dtd *testTargetDelivery) Abort(ctx context.Context) error { return dtd.tgt.AbortErr } @@ -251,6 +285,54 @@ func DoTestDeliveryNonAtomic(t *testing.T, c module.StatusCollector, tgt module. return encodedID } +// DoTestDeliveryStream is similar to DoTestDelivery but passes the body using +// module.StreamingDelivery. +func DoTestDeliveryStream(t *testing.T, tgt module.DeliveryTarget, from string, to []string) string { + t.Helper() + + IDRaw := sha1.Sum([]byte(t.Name())) + encodedID := hex.EncodeToString(IDRaw[:]) + testCtx := context.Background() + + msgMeta := module.MsgMetadata{ + DontTraceSender: true, + ID: encodedID, + OriginalFrom: from, + } + delivery, err := tgt.StartDelivery(testCtx, &msgMeta, from) + if err != nil { + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + for _, rcpt := range to { + if err := delivery.AddRcpt(testCtx, rcpt, smtp.RcptOptions{}); err != nil { + if err := delivery.Abort(testCtx); err != nil { + t.Log("-- delivery.Abort:", err, exterrors.Fields(err)) + } + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + } + + streamDelivery, ok := delivery.(module.StreamingDelivery) + if !ok || !streamDelivery.CanStreamBody() { + t.Fatalf("Delivery object %T can't stream the body", delivery) + } + + hdr := textproto.Header{} + hdr.Add("B", "2") + hdr.Add("A", "1") + if err := streamDelivery.BodyStream(testCtx, hdr, strings.NewReader("foobar\r\n")); err != nil { + if err := delivery.Abort(testCtx); err != nil { + t.Log("-- delivery.Abort:", err, exterrors.Fields(err)) + } + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + if err := delivery.Commit(testCtx); err != nil { + t.Fatalf("Unexpected err: %v %+v", err, exterrors.Fields(err)) + } + + return encodedID +} + const DeliveryData = "A: 1\r\n" + "B: 2\r\n" + "\r\n" + From c79b3a44a1aaa0b14fa59136d37919e93140bb76 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 10:26:48 +0100 Subject: [PATCH 3/9] msgpipeline: Stream the body when nothing in the pipeline reads it The pipeline can pass the body through without buffering only when no check, modifier or DMARC evaluation needs to read it first, and the message goes to exactly one target that can stream. Received and Authentication-Results are still added, since they only touch the header. --- internal/msgpipeline/bodystream_test.go | 128 ++++++++++++++++++++++++ internal/msgpipeline/msgpipeline.go | 76 +++++++++++--- 2 files changed, 191 insertions(+), 13 deletions(-) create mode 100644 internal/msgpipeline/bodystream_test.go diff --git a/internal/msgpipeline/bodystream_test.go b/internal/msgpipeline/bodystream_test.go new file mode 100644 index 00000000..d39cac4d --- /dev/null +++ b/internal/msgpipeline/bodystream_test.go @@ -0,0 +1,128 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package msgpipeline + +import ( + "context" + "testing" + + "github.com/emersion/go-smtp" + "github.com/foxcpp/maddy/framework/module" + "github.com/foxcpp/maddy/internal/modify" + "github.com/foxcpp/maddy/internal/testutils" +) + +func streamTestPipeline(t *testing.T, targets ...module.DeliveryTarget) *MsgPipeline { + return &MsgPipeline{ + msgpipelineCfg: msgpipelineCfg{ + perSource: map[string]sourceBlock{}, + defaultSource: sourceBlock{ + perRcpt: map[string]*rcptBlock{}, + defaultRcpt: &rcptBlock{ + targets: targets, + }, + }, + }, + Log: testutils.Logger(t, "msgpipeline"), + } +} + +func TestMsgPipeline_BodyStream(t *testing.T) { + target := testutils.Target{Streaming: true} + d := streamTestPipeline(t, &target) + + testutils.DoTestDeliveryStream(t, d, "sender@example.com", []string{"rcpt1@example.com", "rcpt2@example.com"}) + + if len(target.Messages) != 1 { + t.Fatalf("wrong amount of messages received, want %d, got %d", 1, len(target.Messages)) + } + testutils.CheckTestMessage(t, &target, 0, "sender@example.com", []string{"rcpt1@example.com", "rcpt2@example.com"}) + if !target.Messages[0].Streamed { + t.Error("message was not streamed") + } +} + +func TestMsgPipeline_CanStreamBody(t *testing.T) { + canStream := func(t *testing.T, d *MsgPipeline) bool { + t.Helper() + + delivery, err := d.StartDelivery(context.Background(), &module.MsgMetadata{ID: "test"}, "sender@example.com") + if err != nil { + t.Fatal(err) + } + defer func() { + if err := delivery.Abort(context.Background()); err != nil { + t.Error(err) + } + }() + if err := delivery.AddRcpt(context.Background(), "rcpt@example.com", smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + return delivery.(module.StreamingDelivery).CanStreamBody() + } + + t.Run("single streaming target", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + if !canStream(t, d) { + t.Error("expected streaming to be possible") + } + }) + t.Run("target without streaming", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{}) + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("two targets", func(t *testing.T) { + d := streamTestPipeline(t, + &testutils.Target{Streaming: true, InstName: "a"}, + &testutils.Target{Streaming: true, InstName: "b"}) + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("global check", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.globalChecks = []module.Check{&testutils.Check{}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("per-recipient check", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.defaultSource.defaultRcpt.checks = []module.Check{&testutils.Check{}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("source modifier", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.defaultSource.modifiers = modify.Group{Modifiers: []module.Modifier{testutils.Modifier{}}} + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) + t.Run("dmarc", func(t *testing.T) { + d := streamTestPipeline(t, &testutils.Target{Streaming: true}) + d.doDMARC = true + if canStream(t, d) { + t.Error("expected streaming to be impossible") + } + }) +} diff --git a/internal/msgpipeline/msgpipeline.go b/internal/msgpipeline/msgpipeline.go index 6025bf06..6fb818cf 100644 --- a/internal/msgpipeline/msgpipeline.go +++ b/internal/msgpipeline/msgpipeline.go @@ -20,6 +20,7 @@ package msgpipeline import ( "context" + "io" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -393,19 +394,7 @@ func (dd *msgpipelineDelivery) Body(ctx context.Context, header textproto.Header } } - if dd.d.FirstPipeline { - // Add Received *after* checks to make sure they see the message literally - // how we received it BUT place it below any other field that might be - // added by applyResults (including Authentication-Results) - // per recommendation in RFC 7001, Section 4 (see GH issue #135). - received, err := target.GenerateReceived(ctx, dd.msgMeta, dd.d.Hostname, dd.msgMeta.OriginalFrom) - if err != nil { - return err - } - header.Add("Received", received) - } - - if err := dd.checkRunner.applyResults(dd.d.Hostname, &header); err != nil { + if err := dd.addTraceFields(ctx, &header); err != nil { return err } @@ -432,6 +421,67 @@ func (dd *msgpipelineDelivery) Body(ctx context.Context, header textproto.Header return nil } +func (dd *msgpipelineDelivery) addTraceFields(ctx context.Context, header *textproto.Header) error { + if dd.d.FirstPipeline { + // Add Received *after* checks to make sure they see the message literally + // how we received it BUT place it below any other field that might be + // added by applyResults (including Authentication-Results) + // per recommendation in RFC 7001, Section 4 (see GH issue #135). + received, err := target.GenerateReceived(ctx, dd.msgMeta, dd.d.Hostname, dd.msgMeta.OriginalFrom) + if err != nil { + return err + } + header.Add("Received", received) + } + + return dd.checkRunner.applyResults(dd.d.Hostname, header) +} + +// CanStreamBody implements module.StreamingDelivery. +// +// The body can be streamed only if nothing in the pipeline needs to read it +// before the target does, and there is exactly one target to read it. +func (dd *msgpipelineDelivery) CanStreamBody() bool { + if dd.checkRunner.doDMARC { + return false + } + if len(dd.d.globalChecks) != 0 || len(dd.sourceBlock.checks) != 0 { + return false + } + if len(dd.d.globalModifiers.Modifiers) != 0 || len(dd.sourceBlock.modifiers.Modifiers) != 0 { + return false + } + for blk := range dd.rcptModifiersState { + if len(blk.checks) != 0 || len(blk.modifiers.Modifiers) != 0 { + return false + } + } + + if len(dd.deliveries) != 1 { + return false + } + var streamDelivery module.StreamingDelivery + for _, delivery := range dd.deliveries { + streamDelivery, _ = delivery.Delivery.(module.StreamingDelivery) + } + return streamDelivery != nil && streamDelivery.CanStreamBody() +} + +// BodyStream implements module.StreamingDelivery. +func (dd *msgpipelineDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + if err := dd.addTraceFields(ctx, &header); err != nil { + return err + } + + for _, delivery := range dd.deliveries { + if err := delivery.Delivery.(module.StreamingDelivery).BodyStream(ctx, header, body); err != nil { + return err + } + dd.log.Debugf("delivery.BodyStream ok, Delivery object = %T", delivery.Delivery) + } + return nil +} + // statusCollector wraps StatusCollector and adds reverse translation // of recipients for all statuses.] // From 14fbb8e96beda2ac23f7c4945907bf548e12edf1 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 10:26:48 +0100 Subject: [PATCH 4/9] endpoint/smtp: Skip body buffering if the pipeline can stream it After reading the header, DATA asks the delivery whether it can stream the body. If so, the body goes straight from the client connection to the target, which for a single target.smtp or target.lmtp means no copy of the message is kept in memory or on disk. If the body fails midway (for example the client disconnects), the delivery is left in place so that the following Reset aborts it. LMTPData keeps buffering, since it relies on per-recipient statuses from BodyNonAtomic. Fixes #882. --- internal/endpoint/smtp/session.go | 57 +++++++++++++++++++++-- internal/endpoint/smtp/smtp_test.go | 72 +++++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 4 deletions(-) diff --git a/internal/endpoint/smtp/session.go b/internal/endpoint/smtp/session.go index 11729151..ead73c61 100644 --- a/internal/endpoint/smtp/session.go +++ b/internal/endpoint/smtp/session.go @@ -428,7 +428,7 @@ func (s *Session) Logout() error { return nil } -func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, error) { +func (s *Session) readHeader(r io.Reader) (textproto.Header, io.Reader, error) { limitr := limitReader(r, s.endp.maxHeaderBytes, &exterrors.SMTPError{ Code: 552, EnhancedCode: exterrors.EnhancedCode{5, 3, 4}, @@ -451,9 +451,26 @@ func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, err // the header size check is done. The message size will be checked by go-smtp limitr.Enabled = false - buf, err := s.endp.buffer(bufr) + return header, bufr, nil +} + +func (s *Session) bufferBody(body io.Reader) (buffer.Buffer, error) { + buf, err := s.endp.buffer(body) if err != nil { - return textproto.Header{}, nil, fmt.Errorf("I/O error while writing buffer: %w", err) + return nil, fmt.Errorf("I/O error while writing buffer: %w", err) + } + return buf, nil +} + +func (s *Session) prepareBody(r io.Reader) (textproto.Header, buffer.Buffer, error) { + header, body, err := s.readHeader(r) + if err != nil { + return textproto.Header{}, nil, err + } + + buf, err := s.bufferBody(body) + if err != nil { + return textproto.Header{}, nil, err } return header, buf, nil @@ -471,7 +488,16 @@ func (s *Session) Data(r io.Reader) error { return s.endp.wrapErr(s.msgMeta.ID, !s.opts.UTF8, "DATA", err) } - header, buf, err := s.prepareBody(r) + header, body, err := s.readHeader(r) + if err != nil { + return wrapErr(err) + } + + if streamDelivery, ok := s.delivery.(module.StreamingDelivery); ok && streamDelivery.CanStreamBody() { + return s.streamBody(bodyCtx, streamDelivery, header, body, wrapErr) + } + + buf, err := s.bufferBody(body) if err != nil { return wrapErr(err) } @@ -501,6 +527,29 @@ func (s *Session) Data(r io.Reader) error { return nil } +// streamBody passes the body to the delivery while it is being read from the +// client. On failure, s.delivery is kept so that Reset aborts it. +func (s *Session) streamBody(ctx context.Context, delivery module.StreamingDelivery, header textproto.Header, body io.Reader, wrapErr func(error) error) error { + if err := s.checkRoutingLoops(header); err != nil { + return wrapErr(err) + } + + if err := delivery.BodyStream(ctx, header, body); err != nil { + return wrapErr(err) + } + + // go-smtp will call Reset, but it will call Abort if delivery is non-nil. + defer s.cleanSession() + + if err := s.delivery.Commit(ctx); err != nil { + return wrapErr(err) + } + + s.log.Msg("accepted", "msg_id", s.msgMeta.ID) + + return nil +} + type statusWrapper struct { sc smtp.StatusCollector s *Session diff --git a/internal/endpoint/smtp/smtp_test.go b/internal/endpoint/smtp/smtp_test.go index 485c7e9e..e5873248 100644 --- a/internal/endpoint/smtp/smtp_test.go +++ b/internal/endpoint/smtp/smtp_test.go @@ -451,6 +451,78 @@ func TestSMTPDelivery_AbortData(t *testing.T) { } } +func TestSMTPDelivery_StreamBody(t *testing.T) { + tgt := testutils.Target{Streaming: true} + endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) + defer func() { + assert.NoError(t, endp.Stop()) + }() + + cl, err := smtp.Dial("127.0.0.1:" + testPort) + if err != nil { + t.Fatal(err) + } + defer func() { + assert.NoError(t, cl.Close()) + }() + + err = submitMsg(t, cl, "sender@example.org", []string{"rcpt1@example.com", "rcpt2@example.com"}, testMsg) + if err != nil { + t.Fatal(err) + } + + if len(tgt.Messages) != 1 { + t.Fatal("Expected a message, got", len(tgt.Messages)) + } + msg := tgt.Messages[0] + if !msg.Streamed { + t.Error("Message body was not streamed") + } + testutils.CheckMsgID(t, &msg, "sender@example.org", []string{"rcpt1@example.com", "rcpt2@example.com"}, "") + if msg.Header.Get("Received") == "" { + t.Error("Missing Received field") + } +} + +func TestSMTPDelivery_StreamBody_AbortData(t *testing.T) { + tgt := testutils.Target{Streaming: true} + endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) + defer func() { + assert.NoError(t, endp.Stop()) + }() + + conn, err := net.Dial("tcp", "127.0.0.1:"+testPort) + if err != nil { + t.Fatal(err) + } + cl := smtp.NewClient(conn) + + if err := cl.Hello("mx.example.org"); err != nil { + t.Fatal(err) + } + if err := cl.Mail("sender@example.org", nil); err != nil { + t.Fatal(err) + } + if err := cl.Rcpt("test@example.com", &smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + if _, err := cl.Data(); err != nil { + t.Fatal(err) + } + // Send the header and a part of the body, then close the connection + // without sending the final dot. + if _, err := conn.Write([]byte(testMsg)); err != nil { + t.Fatal(err) + } + require.NoError(t, conn.Close()) + + time.Sleep(250 * time.Millisecond) + + if len(tgt.Messages) != 0 { + t.Fatal("Expected no messages, got", len(tgt.Messages)) + } +} + func TestSMTPDelivery_EmptyMessage(t *testing.T) { tgt := testutils.Target{} endp := testEndpoint(t, "smtp", nil, &tgt, nil, nil) From ed6fcc5a5b2fe8a0703c51a9ceae4d72bed473bc Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 16:01:06 +0100 Subject: [PATCH 5/9] endpoint/smtp: Share loop check, cleanup and log between body paths checkRoutingLoops only needs the header, so it now runs once right after readHeader for both the buffered and the streamed body. The buffer removal and cleanSession are separate defers, and Commit and the "accepted" log are shared as well. cleanSession is deferred only once the body was passed to the delivery successfully. If Body or BodyStream fails, s.delivery is kept and the Reset that go-smtp calls after DATA aborts it. Before, a failed Body in the buffered path cleaned the session without ever calling Abort. --- internal/endpoint/smtp/session.go | 54 ++++++++++--------------------- 1 file changed, 17 insertions(+), 37 deletions(-) diff --git a/internal/endpoint/smtp/session.go b/internal/endpoint/smtp/session.go index ead73c61..2136f4dc 100644 --- a/internal/endpoint/smtp/session.go +++ b/internal/endpoint/smtp/session.go @@ -493,55 +493,35 @@ func (s *Session) Data(r io.Reader) error { return wrapErr(err) } - if streamDelivery, ok := s.delivery.(module.StreamingDelivery); ok && streamDelivery.CanStreamBody() { - return s.streamBody(bodyCtx, streamDelivery, header, body, wrapErr) - } - - buf, err := s.bufferBody(body) - if err != nil { - return wrapErr(err) - } - defer func() { - if err := buf.Remove(); err != nil { - s.log.Error("failed to remove buffered body", err) - } - - // go-smtp will call Reset, but it will call Abort if delivery is non-nil. - s.cleanSession() - }() - if err := s.checkRoutingLoops(header); err != nil { return wrapErr(err) } - if err := s.delivery.Body(bodyCtx, header, buf); err != nil { - return wrapErr(err) - } - - if err := s.delivery.Commit(bodyCtx); err != nil { - return wrapErr(err) - } - - s.log.Msg("accepted", "msg_id", s.msgMeta.ID) - - return nil -} + if streamDelivery, ok := s.delivery.(module.StreamingDelivery); ok && streamDelivery.CanStreamBody() { + err = streamDelivery.BodyStream(bodyCtx, header, body) + } else { + var buf buffer.Buffer + buf, err = s.bufferBody(body) + if err != nil { + return wrapErr(err) + } + defer func() { + if err := buf.Remove(); err != nil { + s.log.Error("failed to remove buffered body", err) + } + }() -// streamBody passes the body to the delivery while it is being read from the -// client. On failure, s.delivery is kept so that Reset aborts it. -func (s *Session) streamBody(ctx context.Context, delivery module.StreamingDelivery, header textproto.Header, body io.Reader, wrapErr func(error) error) error { - if err := s.checkRoutingLoops(header); err != nil { - return wrapErr(err) + err = s.delivery.Body(bodyCtx, header, buf) } - - if err := delivery.BodyStream(ctx, header, body); err != nil { + if err != nil { + // s.delivery is kept so that Reset aborts it. return wrapErr(err) } // go-smtp will call Reset, but it will call Abort if delivery is non-nil. defer s.cleanSession() - if err := s.delivery.Commit(ctx); err != nil { + if err := s.delivery.Commit(bodyCtx); err != nil { return wrapErr(err) } From cdb1c6c79315b9d2118037022718e7ecc7fb8d7c Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 16:01:55 +0100 Subject: [PATCH 6/9] target/queue: Implement StreamingDelivery The queue reads the body once to write it to the queue directory, so it can take it directly from the message source in all cases. storeNewMessage now takes an io.Reader, and Body opens the buffer before calling it. If reading the body fails, storeNewMessage already removes the partially written files. --- internal/target/queue/queue.go | 42 ++++++++++++++++++--------- internal/target/queue/queue_test.go | 45 +++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 14 deletions(-) diff --git a/internal/target/queue/queue.go b/internal/target/queue/queue.go index 1aec8213..5c7633bf 100644 --- a/internal/target/queue/queue.go +++ b/internal/target/queue/queue.go @@ -600,8 +600,32 @@ func (qd *queueDelivery) AddRcpt(ctx context.Context, rcptTo string, _ smtp.Rcpt func (qd *queueDelivery) Body(ctx context.Context, header textproto.Header, body buffer.Buffer) error { defer trace.StartRegion(ctx, "queue/Body").End() + bodyReader, err := body.Open() + if err != nil { + return err + } + defer func() { + if err := bodyReader.Close(); err != nil { + qd.q.log.Error("bodyReader close failed", err) + } + }() + // Body buffer initially passed to us may not be valid after "delivery" to queue completes. // storeNewMessage returns a new buffer object created from message blob stored on disk. + return qd.store(header, bodyReader) +} + +func (qd *queueDelivery) CanStreamBody() bool { + return true +} + +func (qd *queueDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + defer trace.StartRegion(ctx, "queue/BodyStream").End() + + return qd.store(header, body) +} + +func (qd *queueDelivery) store(header textproto.Header, body io.Reader) error { storedBody, err := qd.q.storeNewMessage(qd.meta, header, body) if err != nil { return err @@ -752,7 +776,7 @@ func (q *Queue) readDiskQueue() error { return nil } -func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, body buffer.Buffer) (buffer.Buffer, error) { +func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, body io.Reader) (buffer.Buffer, error) { id := meta.MsgMeta.ID headerPath := filepath.Join(q.location, id+".header") @@ -771,17 +795,6 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo return nil, err } - bodyReader, err := body.Open() - if err != nil { - q.tryRemoveDanglingFile(id + ".header") - return nil, err - } - defer func() { - if err := bodyReader.Close(); err != nil { - q.log.Error("bodyReader close failed", err) - } - }() - bodyPath := filepath.Join(q.location, id+".body") bodyFile, err := os.Create(bodyPath) if err != nil { @@ -793,7 +806,8 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo } }() - if _, err := io.Copy(bodyFile, bodyReader); err != nil { + bodyLen, err := io.Copy(bodyFile, body) + if err != nil { q.tryRemoveDanglingFile(id + ".body") q.tryRemoveDanglingFile(id + ".header") return nil, err @@ -815,7 +829,7 @@ func (q *Queue) storeNewMessage(meta *QueueMetadata, header textproto.Header, bo queuedMsgs.WithLabelValues(q.name, q.location).Inc() - return buffer.FileBuffer{Path: bodyPath, LenHint: body.Len()}, nil + return buffer.FileBuffer{Path: bodyPath, LenHint: int(bodyLen)}, nil } func (q *Queue) updateMetadataOnDisk(meta *QueueMetadata) error { diff --git a/internal/target/queue/queue_test.go b/internal/target/queue/queue_test.go index 2ad9e821..4c51010d 100644 --- a/internal/target/queue/queue_test.go +++ b/internal/target/queue/queue_test.go @@ -30,6 +30,7 @@ import ( "reflect" "strings" "testing" + "testing/iotest" "time" "github.com/emersion/go-message/textproto" @@ -255,6 +256,50 @@ func TestQueueDelivery(t *testing.T) { checkQueueDir(t, q, []string{}) } +func TestQueueDelivery_BodyStream(t *testing.T) { + t.Parallel() + + dt := unreliableTarget{committed: make(chan testutils.Msg, 10)} + q := newTestQueue(t, &dt) + defer cleanQueue(t, q) + + testutils.DoTestDeliveryStream(t, q, "tester@example.com", []string{"tester1@example.org", "tester2@example.org"}) + + msg := readMsgChanTimeout(t, dt.committed, 5*time.Second) + require.NoError(t, q.Stop()) + + testutils.CheckMsgID(t, msg, "tester@example.com", []string{"tester1@example.org", "tester2@example.org"}, "") + + checkQueueDir(t, q, []string{}) +} + +func TestQueueDelivery_BodyStream_ReadErr(t *testing.T) { + t.Parallel() + + dt := unreliableTarget{ + committed: make(chan testutils.Msg, 10), + aborted: make(chan testutils.Msg, 10), + } + q := newTestQueue(t, &dt) + defer cleanQueue(t, q) + + ctx := context.Background() + delivery, err := q.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "tester@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "tester1@example.org", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + require.NoError(t, delivery.Abort(ctx)) + + require.NoError(t, q.Stop()) + require.Empty(t, dt.committed) + require.Empty(t, dt.aborted) + checkQueueDir(t, q, []string{}) +} + func TestQueueDelivery_PermanentFail_NonPartial(t *testing.T) { t.Parallel() From c5e5553a358e2830ffc983fb2b227eb3a564675c Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 16:03:13 +0100 Subject: [PATCH 7/9] target: Move body read error tracking out of target/smtp Other targets implementing StreamingDelivery need to tell an error from the message source apart from their own errors as well. --- internal/target/smtp/smtp_downstream.go | 19 ++-------- internal/target/source_reader.go | 47 +++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 16 deletions(-) create mode 100644 internal/target/source_reader.go diff --git a/internal/target/smtp/smtp_downstream.go b/internal/target/smtp/smtp_downstream.go index 48926aa5..8fd91060 100644 --- a/internal/target/smtp/smtp_downstream.go +++ b/internal/target/smtp/smtp_downstream.go @@ -308,30 +308,17 @@ func (d *delivery) CanStreamBody() bool { return true } -type errorRecordingReader struct { - r io.Reader - err error -} - -func (r *errorRecordingReader) Read(b []byte) (int, error) { - n, err := r.r.Read(b) - if err != nil && err != io.EOF { - r.err = err - } - return n, err -} - func (d *delivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { - r := &errorRecordingReader{r: body} + r := target.NewSourceReader(body) err := d.conn.Data(ctx, header, r) - if r.err != nil { + if r.Err() != nil { // Sending QUIT would first terminate the DATA command and make the // server accept the partial message. Drop the connection instead. d.connDropped = true if err := d.conn.DirectClose(); err != nil { d.log.Error("failed to close SMTP connection", err) } - return r.err + return r.Err() } return d.u.moduleError(err) } diff --git a/internal/target/source_reader.go b/internal/target/source_reader.go new file mode 100644 index 00000000..42aac1e2 --- /dev/null +++ b/internal/target/source_reader.go @@ -0,0 +1,47 @@ +/* +Maddy Mail Server - Composable all-in-one email server. +Copyright © 2019-2025 Max Mazurov , Maddy Mail Server contributors + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +*/ + +package target + +import "io" + +// SourceReader wraps the body passed to module.StreamingDelivery.BodyStream +// and records the error returned by the message source, so that the target +// can tell it apart from its own errors. +type SourceReader struct { + r io.Reader + err error +} + +func NewSourceReader(r io.Reader) *SourceReader { + return &SourceReader{r: r} +} + +func (r *SourceReader) Read(b []byte) (int, error) { + n, err := r.r.Read(b) + if err != nil && err != io.EOF { + r.err = err + } + return n, err +} + +// Err returns the last error other than io.EOF returned by the underlying +// reader. +func (r *SourceReader) Err() error { + return r.err +} From 36ed55e3204ce643c10526e038379586e0df23a3 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 16:03:23 +0100 Subject: [PATCH 8/9] target/ignore_error: Implement StreamingDelivery The body can be streamed if the wrapped delivery can stream it, or if there is no wrapped delivery because StartDelivery failed. Errors from the wrapped delivery are still ignored, but an error from reading the body comes from the message source and is returned. After a wrapped error the rest of the body is read and discarded, as the buffered path would have, so a read error after that point is not lost. --- internal/target/ignore_error/ignore_error.go | 29 +++++++ .../target/ignore_error/ignore_error_test.go | 87 +++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/internal/target/ignore_error/ignore_error.go b/internal/target/ignore_error/ignore_error.go index 82efc65e..72058431 100644 --- a/internal/target/ignore_error/ignore_error.go +++ b/internal/target/ignore_error/ignore_error.go @@ -33,6 +33,7 @@ package ignore_error import ( "context" + "io" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -116,6 +117,34 @@ func (d *delivery) Body(ctx context.Context, header textproto.Header, body buffe return nil } +func (d *delivery) CanStreamBody() bool { + if d.wrapped == nil { + return true + } + streamDelivery, ok := d.wrapped.(module.StreamingDelivery) + return ok && streamDelivery.CanStreamBody() +} + +// BodyStream implements module.StreamingDelivery. +// +// Errors from the wrapped delivery are ignored as in Body, but errors from +// reading the body are not, since they come from the message source. +func (d *delivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + r := target.NewSourceReader(body) + if d.wrapped != nil { + if err := d.wrapped.(module.StreamingDelivery).BodyStream(ctx, header, r); err != nil && r.Err() == nil { + d.log.Error("ignored error from wrapped target", err) + } + } + + // Consume the rest of the body, as the buffered path does, so that + // the message source can still report a read error. + if r.Err() == nil { + _, _ = io.Copy(io.Discard, r) + } + return r.Err() +} + func (d *delivery) Abort(ctx context.Context) error { if d.wrapped == nil { return nil diff --git a/internal/target/ignore_error/ignore_error_test.go b/internal/target/ignore_error/ignore_error_test.go index 4f2b1897..a78e86ae 100644 --- a/internal/target/ignore_error/ignore_error_test.go +++ b/internal/target/ignore_error/ignore_error_test.go @@ -19,10 +19,17 @@ along with this program. If not, see . package ignore_error import ( + "context" "errors" + "io" + "strings" "testing" + "testing/iotest" + "github.com/emersion/go-message/textproto" + "github.com/emersion/go-smtp" "github.com/foxcpp/maddy/framework/config" + "github.com/foxcpp/maddy/framework/module" "github.com/foxcpp/maddy/internal/testutils" ) @@ -64,6 +71,86 @@ func TestIgnoredErrors(t *testing.T) { } } +func TestBodyStream(t *testing.T) { + inner := &testutils.Target{Streaming: true} + tgt := wrap(t, inner) + + testutils.DoTestDeliveryStream(t, tgt, "from@example.org", []string{"to@example.org"}) + + testutils.CheckTestMessage(t, inner, 0, "from@example.org", []string{"to@example.org"}) + if !inner.Messages[0].Streamed { + t.Error("message body was not streamed to the wrapped target") + } +} + +func TestBodyStreamNotSupported(t *testing.T) { + tgt := wrap(t, &testutils.Target{}) + + d, err := tgt.StartDelivery(context.Background(), &module.MsgMetadata{}, "from@example.org") + if err != nil { + t.Fatal(err) + } + if d.(module.StreamingDelivery).CanStreamBody() { + t.Error("CanStreamBody should be false if the wrapped delivery can't stream") + } +} + +func startStream(t *testing.T, tgt *Target) module.StreamingDelivery { + t.Helper() + + d, err := tgt.StartDelivery(context.Background(), &module.MsgMetadata{}, "from@example.org") + if err != nil { + t.Fatal(err) + } + if err := d.AddRcpt(context.Background(), "to@example.org", smtp.RcptOptions{}); err != nil { + t.Fatal(err) + } + streamDelivery := d.(module.StreamingDelivery) + if !streamDelivery.CanStreamBody() { + t.Fatal("CanStreamBody is false") + } + return streamDelivery +} + +func TestBodyStreamIgnoredError(t *testing.T) { + boom := errors.New("boom") + + for name, inner := range map[string]*testutils.Target{ + "StartDelivery": {Streaming: true, StartErr: boom}, + "BodyStream": {Streaming: true, BodyErr: boom}, + } { + t.Run(name, func(t *testing.T) { + d := startStream(t, wrap(t, inner)) + + body := strings.NewReader("foobar\r\n") + if err := d.BodyStream(context.Background(), textproto.Header{}, body); err != nil { + t.Errorf("wrapped error was not ignored: %v", err) + } + if body.Len() != 0 { + t.Errorf("body was not read to the end, %d bytes left", body.Len()) + } + }) + } +} + +func TestBodyStreamReadError(t *testing.T) { + readErr := errors.New("client went away") + + for name, inner := range map[string]*testutils.Target{ + "Passthrough": {Streaming: true}, + "BodyStream": {Streaming: true, BodyErr: errors.New("boom")}, + } { + t.Run(name, func(t *testing.T) { + d := startStream(t, wrap(t, inner)) + + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + if err := d.BodyStream(context.Background(), textproto.Header{}, body); !errors.Is(err, readErr) { + t.Errorf("expected the read error, got %v", err) + } + }) + } +} + func TestConfigureRequiresTarget(t *testing.T) { tgt := &Target{instName: "test", log: testutils.Logger(t, modName)} From 33b67a9758c69f2d8af5e5ad9f7a00dd8e34cde8 Mon Sep 17 00:00:00 2001 From: Parman Mohammadalizadeh Date: Thu, 29 Oct 2026 16:04:30 +0100 Subject: [PATCH 9/9] target/remote: Implement StreamingDelivery for a single connection The body can be streamed only if it is sent over one connection, that is, all recipients are in one domain. With more connections, each of them needs to read the body, so it is still buffered. If reading the body fails, the connection is dropped without QUIT, as in target/smtp, so the partial message is not accepted, and it is not returned to the pool. closeConn now skips connections that were already dropped. --- internal/target/remote/connect.go | 4 + internal/target/remote/remote.go | 42 +++++++++++ internal/target/remote/remote_test.go | 101 ++++++++++++++++++++++++++ 3 files changed, 147 insertions(+) diff --git a/internal/target/remote/connect.go b/internal/target/remote/connect.go index 3b9692d0..9f71184a 100644 --- a/internal/target/remote/connect.go +++ b/internal/target/remote/connect.go @@ -212,6 +212,10 @@ func (rd *remoteDelivery) closeConn(c *mxConn) { rd.rt.limits.ReleaseDest(c.domain) } + // Client is nil if the connection was already dropped. + if c.Client() == nil { + return + } if err := c.Close(); err != nil { rd.log.Error("client connection close failed", err) } diff --git a/internal/target/remote/remote.go b/internal/target/remote/remote.go index daf5a874..26828722 100644 --- a/internal/target/remote/remote.go +++ b/internal/target/remote/remote.go @@ -28,6 +28,7 @@ import ( "crypto/tls" "errors" "fmt" + "io" "net" "runtime/trace" "strings" @@ -442,6 +443,47 @@ func (rd *remoteDelivery) BodyNonAtomic(ctx context.Context, c module.StatusColl wg.Wait() } +// CanStreamBody implements module.StreamingDelivery. +// +// The body can be streamed only if it is sent over a single connection, +// that is, all recipients are in one domain. +func (rd *remoteDelivery) CanStreamBody() bool { + return len(rd.connections) == 1 +} + +// BodyStream implements module.StreamingDelivery. +func (rd *remoteDelivery) BodyStream(ctx context.Context, header textproto.Header, body io.Reader) error { + defer trace.StartRegion(ctx, "remote/BodyStream").End() + + if rd.msgMeta.Quarantine { + return &exterrors.SMTPError{ + Code: 550, + EnhancedCode: exterrors.EnhancedCode{5, 7, 0}, + Message: "Refusing to deliver quarantined message", + TargetName: "remote", + } + } + + for _, conn := range rd.connections { + r := target.NewSourceReader(body) + err := conn.Data(ctx, header, r) + conn.errored = err != nil + conn.lastUseAt = time.Now() + if r.Err() != nil { + // Sending QUIT would first terminate the DATA command and make the + // server accept the partial message. Drop the connection instead. + if err := conn.DirectClose(); err != nil { + rd.log.Error("failed to close SMTP connection", err) + } + return r.Err() + } + if err != nil { + return err + } + } + return nil +} + func (rd *remoteDelivery) Abort(ctx context.Context) error { return rd.Close() } diff --git a/internal/target/remote/remote_test.go b/internal/target/remote/remote_test.go index 06f0cb80..8e351341 100644 --- a/internal/target/remote/remote_test.go +++ b/internal/target/remote/remote_test.go @@ -21,12 +21,16 @@ package remote import ( "context" "crypto/tls" + "errors" "flag" + "io" "math/rand" "net" "os" "strconv" + "strings" "testing" + "testing/iotest" "github.com/emersion/go-message/textproto" "github.com/emersion/go-smtp" @@ -142,6 +146,103 @@ func TestRemoteDelivery(t *testing.T) { be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid"}) } +func TestRemoteDelivery_BodyStream(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + defer func() { + assert.NoError(t, tgt.Stop()) + }() + testutils.DoTestDeliveryStream(t, tgt, "test@example.com", []string{"test@example.invalid", "test2@example.invalid"}) + + be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid", "test2@example.invalid"}) +} + +func TestRemoteDelivery_BodyStream_MultipleConns(t *testing.T) { + _, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "example2.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + defer func() { + assert.NoError(t, tgt.Stop()) + }() + + ctx := context.Background() + delivery, err := tgt.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "test@example.invalid", smtp.RcptOptions{})) + require.True(t, delivery.(module.StreamingDelivery).CanStreamBody()) + require.NoError(t, delivery.AddRcpt(ctx, "test@example2.invalid", smtp.RcptOptions{})) + require.False(t, delivery.(module.StreamingDelivery).CanStreamBody()) + require.NoError(t, delivery.Abort(ctx)) +} + +func TestRemoteDelivery_BodyStream_ReadErr(t *testing.T) { + be, srv := testutils.SMTPServer(t, "127.0.0.1:"+smtpPort) + defer func() { + require.NoError(t, srv.Close()) + }() + defer testutils.CheckSMTPConnLeak(t, srv) + zones := map[string]mockdns.Zone{ + "example.invalid.": { + MX: []net.MX{{Host: "mx.example.invalid.", Pref: 10}}, + }, + "mx.example.invalid.": { + A: []string{"127.0.0.1"}, + }, + } + + tgt := testTarget(t, zones, nil, nil) + tgt.connReuseLimit = 5 + defer func() { + assert.NoError(t, tgt.Stop()) + }() + + ctx := context.Background() + delivery, err := tgt.StartDelivery(ctx, &module.MsgMetadata{ID: "test"}, "test@example.com") + require.NoError(t, err) + require.NoError(t, delivery.AddRcpt(ctx, "test@example.invalid", smtp.RcptOptions{})) + + readErr := errors.New("client went away") + body := io.MultiReader(strings.NewReader("foo"), iotest.ErrReader(readErr)) + err = delivery.(module.StreamingDelivery).BodyStream(ctx, textproto.Header{}, body) + require.ErrorIs(t, err, readErr) + require.NoError(t, delivery.Abort(ctx)) + + testutils.WaitForConnsClose(t, srv) + require.Empty(t, be.Messages) + + // The dropped connection must not be returned to the pool. + testutils.DoTestDelivery(t, tgt, "test@example.com", []string{"test@example.invalid"}) + be.CheckMsg(t, 0, "test@example.com", []string{"test@example.invalid"}) +} + func TestRemoteDelivery_NoMXFallback(t *testing.T) { tarpit := testutils.FailOnConn(t, "127.0.0.1:"+smtpPort) defer func() {