1// Package queue is in charge of outgoing messages, queueing them when submitted,
2// attempting a first delivery over SMTP, retrying with backoff and sending DSNs
3// for delayed or failed deliveries.
21 "golang.org/x/net/proxy"
23 "github.com/prometheus/client_golang/prometheus"
24 "github.com/prometheus/client_golang/prometheus/promauto"
26 "github.com/mjl-/bstore"
28 "github.com/mjl-/mox/config"
29 "github.com/mjl-/mox/dns"
30 "github.com/mjl-/mox/dsn"
31 "github.com/mjl-/mox/metrics"
32 "github.com/mjl-/mox/mlog"
33 "github.com/mjl-/mox/mox-"
34 "github.com/mjl-/mox/moxio"
35 "github.com/mjl-/mox/moxvar"
36 "github.com/mjl-/mox/smtp"
37 "github.com/mjl-/mox/smtpclient"
38 "github.com/mjl-/mox/store"
39 "github.com/mjl-/mox/tlsrpt"
40 "github.com/mjl-/mox/tlsrptdb"
41 "github.com/mjl-/mox/webapi"
42 "github.com/mjl-/mox/webhook"
45// ErrFromID indicate a fromid was present when adding a message to the queue, but
47var ErrFromID = errors.New("fromid not unique")
50 metricConnection = promauto.NewCounterVec(
51 prometheus.CounterOpts{
52 Name: "mox_queue_connection_total",
53 Help: "Queue client connections, outgoing.",
56 "result", // "ok", "timeout", "canceled", "error"
59 metricDelivery = promauto.NewHistogramVec(
60 prometheus.HistogramOpts{
61 Name: "mox_queue_delivery_duration_seconds",
62 Help: "SMTP client delivery attempt to single host.",
63 Buckets: []float64{0.01, 0.05, 0.100, 0.5, 1, 5, 10, 20, 30, 60, 120},
66 "attempt", // Number of attempts.
67 "transport", // empty for default direct delivery.
68 "tlsmode", // immediate, requiredstarttls, opportunistic, skip (from smtpclient.TLSMode), with optional +mtasts and/or +dane.
69 "result", // ok, timeout, canceled, temperror, permerror, error
72 metricHold = promauto.NewGauge(
74 Name: "mox_queue_hold",
75 Help: "Messages in queue that are on hold.",
80var jitter = mox.NewPseudoRand()
82var DBTypes = []any{Msg{}, HoldRule{}, MsgRetired{}, webapi.Suppression{}, Hook{}, HookRetired{}} // Types stored in DB.
83var DB *bstore.DB // Exported for making backups.
85// Allow requesting delivery starting from up to this interval from time of submission.
86const FutureReleaseIntervalMax = 60 * 24 * time.Hour
88// Set for mox localserve, to prevent queueing.
91// HoldRule is a set of conditions that cause a matching message to be marked as on
92// hold when it is queued. All-empty conditions matches all messages, effectively
93// pausing the entire queue.
97 SenderDomain dns.Domain
98 RecipientDomain dns.Domain
99 SenderDomainStr string // Unicode.
100 RecipientDomainStr string // Unicode.
103func (pr HoldRule) All() bool {
105 return pr == HoldRule{}
108func (pr HoldRule) matches(m Msg) bool {
109 return pr.All() || pr.Account == m.SenderAccount || pr.SenderDomainStr == m.SenderDomainStr || pr.RecipientDomainStr == m.RecipientDomainStr
112// Msg is a message in the queue.
114// Use MakeMsg to make a message with fields that Add needs. Add will further set
115// queueing related fields.
119 // A message for multiple recipients will get a BaseID that is identical to the
120 // first Msg.ID queued. The message contents will be identical for each recipient,
121 // including MsgPrefix. If other properties are identical too, including recipient
122 // domain, multiple Msgs may be delivered in a single SMTP transaction. For
123 // messages with a single recipient, this field will be 0.
124 BaseID int64 `bstore:"index"`
126 Queued time.Time `bstore:"default now"`
127 Hold bool // If set, delivery won't be attempted.
128 SenderAccount string // Failures are delivered back to this local account. Also used for routing.
129 SenderLocalpart smtp.Localpart // Should be a local user and domain.
130 SenderDomain dns.IPDomain
131 SenderDomainStr string // For filtering, unicode.
132 FromID string // For transactional messages, used to match later DSNs.
133 RecipientLocalpart smtp.Localpart // Typically a remote user and domain.
134 RecipientDomain dns.IPDomain
135 RecipientDomainStr string // For filtering, unicode domain. Can also contain ip enclosed in [].
136 Attempts int // Next attempt is based on last attempt and exponential back off based on attempts.
137 MaxAttempts int // Max number of attempts before giving up. If 0, then the default of 8 attempts is used instead.
138 DialedIPs map[string][]net.IP // For each host, the IPs that were dialed. Used for IP selection for later attempts.
139 NextAttempt time.Time // For scheduling.
140 LastAttempt *time.Time
143 Has8bit bool // Whether message contains bytes with high bit set, determines whether 8BITMIME SMTP extension is needed.
144 SMTPUTF8 bool // Whether message requires use of SMTPUTF8.
145 IsDMARCReport bool // Delivery failures for DMARC reports are handled differently.
146 IsTLSReport bool // Delivery failures for TLS reports are handled differently.
147 Size int64 // Full size of message, combined MsgPrefix with contents of message file.
148 MessageID string // Message-ID header, including <>. Used when composing a DSN, in its References header.
149 MsgPrefix []byte // Data to send before the contents from the file, typically with headers like DKIM-Signature.
150 Subject string // For context about delivery.
152 // If set, this message is a DSN and this is a version using utf-8, for the case
153 // the remote MTA supports smtputf8. In this case, Size and MsgPrefix are not
157 // If non-empty, the transport to use for this message. Can be set through cli or
158 // admin interface. If empty (the default for a submitted message), regular routing
162 // RequireTLS influences TLS verification during delivery.
164 // If nil, the recipient domain policy is followed (MTA-STS and/or DANE), falling
165 // back to optional opportunistic non-verified STARTTLS.
167 // If RequireTLS is true (through SMTP REQUIRETLS extension or webmail submit),
168 // MTA-STS or DANE is required, as well as REQUIRETLS support by the next hop
171 // If RequireTLS is false (through messag header "TLS-Required: No"), the recipient
172 // domain's policy is ignored if it does not lead to a successful TLS connection,
173 // i.e. falling back to SMTP delivery with unverified STARTTLS or plain text.
177 // For DSNs, where the original FUTURERELEASE value must be included as per-message
178 // field. This field should be of the form "for;" plus interval, or "until;" plus
180 FutureReleaseRequest string
183 Extra map[string]string // Extra information, for transactional email.
186// MsgResult is the result (or work in progress) of a delivery attempt.
187type MsgResult struct {
189 Duration time.Duration
194 // todo: store smtp trace for failed deliveries for debugging, perhaps also for successful deliveries.
197// Stored in MsgResult.Error while delivery is in progress. Replaced after success/error.
198const resultErrorDelivering = "delivering..."
200// markResult updates/adds a delivery result.
201func (m *Msg) markResult(code int, secode string, errmsg string, success bool) {
202 if len(m.Results) == 0 || m.Results[len(m.Results)-1].Error != resultErrorDelivering {
203 m.Results = append(m.Results, MsgResult{Start: time.Now()})
205 result := &m.Results[len(m.Results)-1]
206 result.Duration = time.Since(result.Start)
208 result.Secode = secode
209 result.Error = errmsg
210 result.Success = success
213// LastResult returns the last result entry, or an empty result.
214func (m *Msg) LastResult() MsgResult {
215 if len(m.Results) == 0 {
216 return MsgResult{Start: time.Now()}
218 return m.Results[len(m.Results)-1]
221// Sender of message as used in MAIL FROM.
222func (m Msg) Sender() smtp.Path {
223 return smtp.Path{Localpart: m.SenderLocalpart, IPDomain: m.SenderDomain}
226// Recipient of message as used in RCPT TO.
227func (m Msg) Recipient() smtp.Path {
228 return smtp.Path{Localpart: m.RecipientLocalpart, IPDomain: m.RecipientDomain}
231// MessagePath returns the path where the message is stored.
232func (m Msg) MessagePath() string {
233 return mox.DataDirPath(filepath.Join("queue", store.MessagePath(m.ID)))
236// todo: store which transport (if any) was actually used in MsgResult, based on routes.
238// Retired returns a MsgRetired for the message, for history of deliveries.
239func (m Msg) Retired(success bool, t, keepUntil time.Time) MsgRetired {
244 SenderAccount: m.SenderAccount,
245 SenderLocalpart: m.SenderLocalpart,
246 SenderDomainStr: m.SenderDomainStr,
248 RecipientLocalpart: m.RecipientLocalpart,
249 RecipientDomain: m.RecipientDomain,
250 RecipientDomainStr: m.RecipientDomainStr,
251 Attempts: m.Attempts,
252 MaxAttempts: m.MaxAttempts,
253 DialedIPs: m.DialedIPs,
254 LastAttempt: m.LastAttempt,
257 SMTPUTF8: m.SMTPUTF8,
258 IsDMARCReport: m.IsDMARCReport,
259 IsTLSReport: m.IsTLSReport,
261 MessageID: m.MessageID,
263 Transport: m.Transport,
264 RequireTLS: m.RequireTLS,
265 FutureReleaseRequest: m.FutureReleaseRequest,
268 RecipientAddress: smtp.Path{Localpart: m.RecipientLocalpart, IPDomain: m.RecipientDomain}.XString(true),
271 KeepUntil: keepUntil,
275// MsgRetired is a message for which delivery completed, either successful,
276// failed/canceled. Retired messages are only stored if so configured, and will be
277// cleaned up after the configured period.
278type MsgRetired struct {
279 ID int64 // Same ID as it was as Msg.ID.
283 SenderAccount string // Failures are delivered back to this local account. Also used for routing.
284 SenderLocalpart smtp.Localpart // Should be a local user and domain.
285 SenderDomainStr string // For filtering, unicode.
286 FromID string `bstore:"index"` // Used to match DSNs.
287 RecipientLocalpart smtp.Localpart // Typically a remote user and domain.
288 RecipientDomain dns.IPDomain
289 RecipientDomainStr string // For filtering, unicode.
290 Attempts int // Next attempt is based on last attempt and exponential back off based on attempts.
291 MaxAttempts int // Max number of attempts before giving up. If 0, then the default of 8 attempts is used instead.
292 DialedIPs map[string][]net.IP // For each host, the IPs that were dialed. Used for IP selection for later attempts.
293 LastAttempt *time.Time
296 Has8bit bool // Whether message contains bytes with high bit set, determines whether 8BITMIME SMTP extension is needed.
297 SMTPUTF8 bool // Whether message requires use of SMTPUTF8.
298 IsDMARCReport bool // Delivery failures for DMARC reports are handled differently.
299 IsTLSReport bool // Delivery failures for TLS reports are handled differently.
300 Size int64 // Full size of message, combined MsgPrefix with contents of message file.
301 MessageID string // Used when composing a DSN, in its References header.
302 Subject string // For context about delivery.
306 FutureReleaseRequest string
308 Extra map[string]string // Extra information, for transactional email.
310 LastActivity time.Time `bstore:"index"`
311 RecipientAddress string `bstore:"index RecipientAddress+LastActivity"`
312 Success bool // Whether delivery to next hop succeeded.
313 KeepUntil time.Time `bstore:"index"`
316// Sender of message as used in MAIL FROM.
317func (m MsgRetired) Sender() (path smtp.Path, err error) {
318 path.Localpart = m.RecipientLocalpart
319 if strings.HasPrefix(m.SenderDomainStr, "[") && strings.HasSuffix(m.SenderDomainStr, "]") {
320 s := m.SenderDomainStr[1 : len(m.SenderDomainStr)-1]
321 path.IPDomain.IP = net.ParseIP(s)
322 if path.IPDomain.IP == nil {
323 err = fmt.Errorf("parsing ip address %q: %v", s, err)
326 path.IPDomain.Domain, err = dns.ParseDomain(m.SenderDomainStr)
331// Recipient of message as used in RCPT TO.
332func (m MsgRetired) Recipient() smtp.Path {
333 return smtp.Path{Localpart: m.RecipientLocalpart, IPDomain: m.RecipientDomain}
336// LastResult returns the last result entry, or an empty result.
337func (m MsgRetired) LastResult() MsgResult {
338 if len(m.Results) == 0 {
341 return m.Results[len(m.Results)-1]
344// Init opens the queue database without starting delivery.
346 qpath := mox.DataDirPath(filepath.FromSlash("queue/index.db"))
347 os.MkdirAll(filepath.Dir(qpath), 0770)
349 if _, err := os.Stat(qpath); err != nil && os.IsNotExist(err) {
354 log := mlog.New("queue", nil)
355 opts := bstore.Options{Timeout: 5 * time.Second, Perm: 0660, RegisterLogger: moxvar.RegisterLogger(qpath, log.Logger)}
356 DB, err = bstore.Open(mox.Shutdown, qpath, &opts, DBTypes...)
358 err = DB.Read(mox.Shutdown, func(tx *bstore.Tx) error {
359 return metricHoldUpdate(tx)
364 err := os.Remove(qpath)
365 log.Check(err, "removing new queue database file after error")
367 return fmt.Errorf("open queue database: %s", err)
372// When we update the gauge, we just get the full current value, not try to account
374func metricHoldUpdate(tx *bstore.Tx) error {
375 count, err := bstore.QueryTx[Msg](tx).FilterNonzero(Msg{Hold: true}).Count()
377 return fmt.Errorf("querying messages on hold for metric: %v", err)
379 metricHold.Set(float64(count))
383// Shutdown closes the queue database. The delivery process isn't stopped. For tests only.
387 mlog.New("queue", nil).Errorx("closing queue db", err)
392// todo: the filtering & sorting can use improvements. too much duplicated code (variants between {Msg,Hook}{,Retired}. Sort has pagination fields, some untyped.
394// Filter filters messages to list or operate on. Used by admin web interface
397// Only non-empty/non-zero values are applied to the filter. Leaving all fields
398// empty/zero matches all messages.
406 Submitted string // Whether submitted before/after a time relative to now. ">$duration" or "<$duration", also with "now" for duration.
407 NextAttempt string // ">$duration" or "<$duration", also with "now" for duration.
411func (f Filter) apply(q *bstore.Query[Msg]) error {
415 applyTime := func(field string, s string) error {
418 if strings.HasPrefix(s, "<") {
420 } else if !strings.HasPrefix(s, ">") {
421 return fmt.Errorf(`must start with "<" for before or ">" for after a duration`)
423 s = strings.TrimSpace(s[1:])
427 } else if d, err := time.ParseDuration(s); err != nil {
428 return fmt.Errorf("parsing duration %q: %v", orig, err)
430 t = time.Now().Add(d)
433 q.FilterLess(field, t)
435 q.FilterGreater(field, t)
440 q.FilterEqual("Hold", *f.Hold)
442 if f.Submitted != "" {
443 if err := applyTime("Queued", f.Submitted); err != nil {
444 return fmt.Errorf("applying filter for submitted: %v", err)
447 if f.NextAttempt != "" {
448 if err := applyTime("NextAttempt", f.NextAttempt); err != nil {
449 return fmt.Errorf("applying filter for next attempt: %v", err)
453 q.FilterNonzero(Msg{SenderAccount: f.Account})
455 if f.Transport != nil {
456 q.FilterEqual("Transport", *f.Transport)
458 if f.From != "" || f.To != "" {
459 q.FilterFn(func(m Msg) bool {
460 return f.From != "" && strings.Contains(m.Sender().XString(true), f.From) || f.To != "" && strings.Contains(m.Recipient().XString(true), f.To)
470 Field string // "Queued" or "NextAttempt"/"".
471 LastID int64 // If > 0, we return objects beyond this, less/greater depending on Asc.
472 Last any // Value of Field for last object. Must be set iff LastID is set.
473 Asc bool // Ascending, or descending.
476func (s Sort) apply(q *bstore.Query[Msg]) error {
478 case "", "NextAttempt":
479 s.Field = "NextAttempt"
483 return fmt.Errorf("unknown sort order field %q", s.Field)
487 ls, ok := s.Last.(string)
489 return fmt.Errorf("last should be string with time, not %T %q", s.Last, s.Last)
491 last, err := time.Parse(time.RFC3339Nano, ls)
493 last, err = time.Parse(time.RFC3339, ls)
496 return fmt.Errorf("parsing last %q as time: %v", s.Last, err)
498 q.FilterNotEqual("ID", s.LastID)
499 var fieldEqual func(m Msg) bool
500 if s.Field == "NextAttempt" {
501 fieldEqual = func(m Msg) bool { return m.NextAttempt.Equal(last) }
503 fieldEqual = func(m Msg) bool { return m.Queued.Equal(last) }
506 q.FilterGreaterEqual(s.Field, last)
507 q.FilterFn(func(m Msg) bool {
508 return !fieldEqual(m) || m.ID > s.LastID
511 q.FilterLessEqual(s.Field, last)
512 q.FilterFn(func(m Msg) bool {
513 return !fieldEqual(m) || m.ID < s.LastID
518 q.SortAsc(s.Field, "ID")
520 q.SortDesc(s.Field, "ID")
525// List returns max 100 messages matching filter in the delivery queue.
526// By default, orders by next delivery attempt.
527func List(ctx context.Context, filter Filter, sort Sort) ([]Msg, error) {
528 q := bstore.QueryDB[Msg](ctx, DB)
529 if err := filter.apply(q); err != nil {
532 if err := sort.apply(q); err != nil {
535 qmsgs, err := q.List()
542// Count returns the number of messages in the delivery queue.
543func Count(ctx context.Context) (int, error) {
544 return bstore.QueryDB[Msg](ctx, DB).Count()
547// HoldRuleList returns all hold rules.
548func HoldRuleList(ctx context.Context) ([]HoldRule, error) {
549 return bstore.QueryDB[HoldRule](ctx, DB).List()
552// HoldRuleAdd adds a new hold rule causing newly submitted messages to be marked
553// as "on hold", and existing matching messages too.
554func HoldRuleAdd(ctx context.Context, log mlog.Log, hr HoldRule) (HoldRule, error) {
556 err := DB.Write(ctx, func(tx *bstore.Tx) error {
558 hr.SenderDomainStr = hr.SenderDomain.Name()
559 hr.RecipientDomainStr = hr.RecipientDomain.Name()
560 if err := tx.Insert(&hr); err != nil {
563 log.Info("adding hold rule", slog.Any("holdrule", hr))
565 q := bstore.QueryTx[Msg](tx)
568 SenderAccount: hr.Account,
569 SenderDomainStr: hr.SenderDomainStr,
570 RecipientDomainStr: hr.RecipientDomainStr,
574 n, err = q.UpdateField("Hold", true)
576 return fmt.Errorf("marking existing matching messages in queue on hold: %v", err)
578 return metricHoldUpdate(tx)
581 return HoldRule{}, err
583 log.Info("marked messages in queue as on hold", slog.Int("messages", n))
588// HoldRuleRemove removes a hold rule. The Hold field of existing messages are not
590func HoldRuleRemove(ctx context.Context, log mlog.Log, holdRuleID int64) error {
591 return DB.Write(ctx, func(tx *bstore.Tx) error {
592 hr := HoldRule{ID: holdRuleID}
593 if err := tx.Get(&hr); err != nil {
596 log.Info("removing hold rule", slog.Any("holdrule", hr))
597 return tx.Delete(HoldRule{ID: holdRuleID})
601// MakeMsg is a convenience function that sets the commonly used fields for a Msg.
602// messageID should include <>.
603func MakeMsg(sender, recipient smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, prefix []byte, requireTLS *bool, next time.Time, subject string) Msg {
605 SenderLocalpart: sender.Localpart,
606 SenderDomain: sender.IPDomain,
607 RecipientLocalpart: recipient.Localpart,
608 RecipientDomain: recipient.IPDomain,
612 MessageID: messageID,
615 RequireTLS: requireTLS,
621// Add one or more new messages to the queue. If the sender paths and MsgPrefix are
622// identical, they'll get the same BaseID, so they can be delivered in a single
623// SMTP transaction, with a single DATA command, but may be split into multiple
624// transactions if errors/limits are encountered. The queue is kicked immediately
625// to start a first delivery attempt.
627// ID of the messagse must be 0 and will be set after inserting in the queue.
629// Add sets derived fields like SenderDomainStr and RecipientDomainStr, and fields
630// related to queueing, such as Queued, NextAttempt.
631func Add(ctx context.Context, log mlog.Log, senderAccount string, msgFile *os.File, qml ...Msg) error {
633 return fmt.Errorf("must queue at least one message")
638 for i, qm := range qml {
640 return fmt.Errorf("id of queued messages must be 0")
642 // Sanity check, internal consistency.
643 qml[i].SenderDomainStr = formatIPDomain(qm.SenderDomain)
644 qml[i].RecipientDomainStr = formatIPDomain(qm.RecipientDomain)
645 if base && i > 0 && qm.Sender().String() != qml[0].Sender().String() || !bytes.Equal(qm.MsgPrefix, qml[0].MsgPrefix) {
650 tx, err := DB.Begin(ctx, true)
652 return fmt.Errorf("begin transaction: %w", err)
656 if err := tx.Rollback(); err != nil {
657 log.Errorx("rollback for queue", err)
662 // Mark messages Hold if they match a hold rule.
663 holdRules, err := bstore.QueryTx[HoldRule](tx).List()
665 return fmt.Errorf("getting queue hold rules")
668 // Insert messages into queue. If multiple messages are to be delivered in a single
669 // transaction, they all get a non-zero BaseID that is the Msg.ID of the first
673 // FromIDs must be unique if present. We don't have a unique index because values
674 // can be the empty string. We check in both Msg and MsgRetired, both are relevant
675 // for uniquely identifying a message sent in the past.
676 if fromID := qml[i].FromID; fromID != "" {
677 if exists, err := bstore.QueryTx[Msg](tx).FilterNonzero(Msg{FromID: fromID}).Exists(); err != nil {
678 return fmt.Errorf("looking up fromid: %v", err)
680 return fmt.Errorf("%w: fromid %q already present in message queue", ErrFromID, fromID)
682 if exists, err := bstore.QueryTx[MsgRetired](tx).FilterNonzero(MsgRetired{FromID: fromID}).Exists(); err != nil {
683 return fmt.Errorf("looking up fromid: %v", err)
685 return fmt.Errorf("%w: fromid %q already present in retired message queue", ErrFromID, fromID)
689 qml[i].SenderAccount = senderAccount
690 qml[i].BaseID = baseID
691 for _, hr := range holdRules {
692 if hr.matches(qml[i]) {
697 if err := tx.Insert(&qml[i]); err != nil {
700 if base && i == 0 && len(qml) > 1 {
702 qml[i].BaseID = baseID
703 if err := tx.Update(&qml[i]); err != nil {
711 for _, p := range paths {
713 log.Check(err, "removing destination message file for queue", slog.String("path", p))
717 syncDirs := map[string]struct{}{}
719 for _, qm := range qml {
720 dst := qm.MessagePath()
721 paths = append(paths, dst)
723 dstDir := filepath.Dir(dst)
724 if _, ok := syncDirs[dstDir]; !ok {
725 os.MkdirAll(dstDir, 0770)
726 syncDirs[dstDir] = struct{}{}
729 if err := moxio.LinkOrCopy(log, dst, msgFile.Name(), nil, true); err != nil {
730 return fmt.Errorf("linking/copying message to new file: %s", err)
734 for dir := range syncDirs {
735 if err := moxio.SyncDir(log, dir); err != nil {
736 return fmt.Errorf("sync directory: %v", err)
740 for _, m := range qml {
742 if err := metricHoldUpdate(tx); err != nil {
749 if err := tx.Commit(); err != nil {
750 return fmt.Errorf("commit transaction: %s", err)
760func formatIPDomain(d dns.IPDomain) string {
762 return "[" + d.IP.String() + "]"
764 return d.Domain.Name()
768 msgqueue = make(chan struct{}, 1)
769 deliveryResults = make(chan string, 1)
779 case msgqueue <- struct{}{}:
784// NextAttemptAdd adds a duration to the NextAttempt for all matching messages, and
786func NextAttemptAdd(ctx context.Context, filter Filter, d time.Duration) (affected int, err error) {
787 err = DB.Write(ctx, func(tx *bstore.Tx) error {
788 q := bstore.QueryTx[Msg](tx)
789 if err := filter.apply(q); err != nil {
792 msgs, err := q.List()
794 return fmt.Errorf("listing matching messages: %v", err)
796 for _, m := range msgs {
797 m.NextAttempt = m.NextAttempt.Add(d)
798 if err := tx.Update(&m); err != nil {
812// NextAttemptSet sets NextAttempt for all matching messages to a new time, and
814func NextAttemptSet(ctx context.Context, filter Filter, t time.Time) (affected int, err error) {
815 q := bstore.QueryDB[Msg](ctx, DB)
816 if err := filter.apply(q); err != nil {
819 n, err := q.UpdateNonzero(Msg{NextAttempt: t})
821 return 0, fmt.Errorf("selecting and updating messages in queue: %v", err)
827// HoldSet sets Hold for all matching messages and kicks the queue.
828func HoldSet(ctx context.Context, filter Filter, hold bool) (affected int, err error) {
829 err = DB.Write(ctx, func(tx *bstore.Tx) error {
830 q := bstore.QueryTx[Msg](tx)
831 if err := filter.apply(q); err != nil {
834 n, err := q.UpdateFields(map[string]any{"Hold": hold})
836 return fmt.Errorf("selecting and updating messages in queue: %v", err)
839 return metricHoldUpdate(tx)
848// TransportSet changes the transport to use for the matching messages.
849func TransportSet(ctx context.Context, filter Filter, transport string) (affected int, err error) {
850 q := bstore.QueryDB[Msg](ctx, DB)
851 if err := filter.apply(q); err != nil {
854 n, err := q.UpdateFields(map[string]any{"Transport": transport})
856 return 0, fmt.Errorf("selecting and updating messages in queue: %v", err)
862// Fail marks matching messages as failed for delivery, delivers a DSN to the
863// sender, and sends a webhook.
865// Returns number of messages removed, which can be non-zero even in case of an
867func Fail(ctx context.Context, log mlog.Log, f Filter) (affected int, err error) {
868 return failDrop(ctx, log, f, true)
871// Drop removes matching messages from the queue. Messages are added as retired
872// message, webhooks with the "canceled" event are queued.
874// Returns number of messages removed, which can be non-zero even in case of an
876func Drop(ctx context.Context, log mlog.Log, f Filter) (affected int, err error) {
877 return failDrop(ctx, log, f, false)
880func failDrop(ctx context.Context, log mlog.Log, filter Filter, fail bool) (affected int, err error) {
882 err = DB.Write(ctx, func(tx *bstore.Tx) error {
883 q := bstore.QueryTx[Msg](tx)
884 if err := filter.apply(q); err != nil {
890 return fmt.Errorf("getting messages to delete: %v", err)
898 var remoteMTA dsn.NameIP
899 for i := range msgs {
902 Error: "delivery canceled by admin",
904 msgs[i].Results = append(msgs[i].Results, result)
906 if msgs[i].LastAttempt == nil {
907 msgs[i].LastAttempt = &now
909 deliverDSNFailure(log, msgs[i], remoteMTA, "", result.Error, nil)
912 event := webhook.EventCanceled
914 event = webhook.EventFailed
916 if err := retireMsgs(log, tx, event, 0, "", nil, msgs...); err != nil {
917 return fmt.Errorf("removing queue messages from database: %w", err)
919 return metricHoldUpdate(tx)
925 if err := removeMsgsFS(log, msgs...); err != nil {
926 return len(msgs), fmt.Errorf("removing queue messages from file system: %w", err)
930 return len(msgs), nil
933// RequireTLSSet updates the RequireTLS field of matching messages.
934func RequireTLSSet(ctx context.Context, filter Filter, requireTLS *bool) (affected int, err error) {
935 q := bstore.QueryDB[Msg](ctx, DB)
936 if err := filter.apply(q); err != nil {
939 n, err := q.UpdateFields(map[string]any{"RequireTLS": requireTLS})
944// RetiredFilter filters messages to list or operate on. Used by admin web interface
947// Only non-empty/non-zero values are applied to the filter. Leaving all fields
948// empty/zero matches all messages.
949type RetiredFilter struct {
955 Submitted string // Whether submitted before/after a time relative to now. ">$duration" or "<$duration", also with "now" for duration.
956 LastActivity string // ">$duration" or "<$duration", also with "now" for duration.
961func (f RetiredFilter) apply(q *bstore.Query[MsgRetired]) error {
965 applyTime := func(field string, s string) error {
968 if strings.HasPrefix(s, "<") {
970 } else if !strings.HasPrefix(s, ">") {
971 return fmt.Errorf(`must start with "<" for before or ">" for after a duration`)
973 s = strings.TrimSpace(s[1:])
977 } else if d, err := time.ParseDuration(s); err != nil {
978 return fmt.Errorf("parsing duration %q: %v", orig, err)
980 t = time.Now().Add(d)
983 q.FilterLess(field, t)
985 q.FilterGreater(field, t)
989 if f.Submitted != "" {
990 if err := applyTime("Queued", f.Submitted); err != nil {
991 return fmt.Errorf("applying filter for submitted: %v", err)
994 if f.LastActivity != "" {
995 if err := applyTime("LastActivity", f.LastActivity); err != nil {
996 return fmt.Errorf("applying filter for last activity: %v", err)
1000 q.FilterNonzero(MsgRetired{SenderAccount: f.Account})
1002 if f.Transport != nil {
1003 q.FilterEqual("Transport", *f.Transport)
1005 if f.From != "" || f.To != "" {
1006 q.FilterFn(func(m MsgRetired) bool {
1007 return f.From != "" && strings.Contains(m.SenderLocalpart.String()+"@"+m.SenderDomainStr, f.From) || f.To != "" && strings.Contains(m.Recipient().XString(true), f.To)
1010 if f.Success != nil {
1011 q.FilterEqual("Success", *f.Success)
1019type RetiredSort struct {
1020 Field string // "Queued" or "LastActivity"/"".
1021 LastID int64 // If > 0, we return objects beyond this, less/greater depending on Asc.
1022 Last any // Value of Field for last object. Must be set iff LastID is set.
1023 Asc bool // Ascending, or descending.
1026func (s RetiredSort) apply(q *bstore.Query[MsgRetired]) error {
1028 case "", "LastActivity":
1029 s.Field = "LastActivity"
1033 return fmt.Errorf("unknown sort order field %q", s.Field)
1037 ls, ok := s.Last.(string)
1039 return fmt.Errorf("last should be string with time, not %T %q", s.Last, s.Last)
1041 last, err := time.Parse(time.RFC3339Nano, ls)
1043 last, err = time.Parse(time.RFC3339, ls)
1046 return fmt.Errorf("parsing last %q as time: %v", s.Last, err)
1048 q.FilterNotEqual("ID", s.LastID)
1049 var fieldEqual func(m MsgRetired) bool
1050 if s.Field == "LastActivity" {
1051 fieldEqual = func(m MsgRetired) bool { return m.LastActivity.Equal(last) }
1053 fieldEqual = func(m MsgRetired) bool { return m.Queued.Equal(last) }
1056 q.FilterGreaterEqual(s.Field, last)
1057 q.FilterFn(func(mr MsgRetired) bool {
1058 return !fieldEqual(mr) || mr.ID > s.LastID
1061 q.FilterLessEqual(s.Field, last)
1062 q.FilterFn(func(mr MsgRetired) bool {
1063 return !fieldEqual(mr) || mr.ID < s.LastID
1068 q.SortAsc(s.Field, "ID")
1070 q.SortDesc(s.Field, "ID")
1075// RetiredList returns retired messages.
1076func RetiredList(ctx context.Context, filter RetiredFilter, sort RetiredSort) ([]MsgRetired, error) {
1077 q := bstore.QueryDB[MsgRetired](ctx, DB)
1078 if err := filter.apply(q); err != nil {
1081 if err := sort.apply(q); err != nil {
1087type ReadReaderAtCloser interface {
1092// OpenMessage opens a message present in the queue.
1093func OpenMessage(ctx context.Context, id int64) (ReadReaderAtCloser, error) {
1095 err := DB.Get(ctx, &qm)
1099 f, err := os.Open(qm.MessagePath())
1101 return nil, fmt.Errorf("open message file: %s", err)
1103 r := store.FileMsgReader(qm.MsgPrefix, f)
1107const maxConcurrentDeliveries = 10
1108const maxConcurrentHookDeliveries = 10
1110// Start opens the database by calling Init, then starts the delivery and cleanup
1112func Start(resolver dns.Resolver, done chan struct{}) error {
1113 if err := Init(); err != nil {
1117 go startQueue(resolver, done)
1118 go startHookQueue(done)
1120 go cleanupMsgRetired(done)
1121 go cleanupHookRetired(done)
1126func cleanupMsgRetired(done chan struct{}) {
1127 log := mlog.New("queue", nil)
1132 log.Error("unhandled panic in cleanupMsgRetired", slog.Any("x", x))
1134 metrics.PanicInc(metrics.Queue)
1138 timer := time.NewTimer(3 * time.Second)
1141 case <-mox.Shutdown.Done():
1147 cleanupMsgRetiredSingle(log)
1148 timer.Reset(time.Hour)
1152func cleanupMsgRetiredSingle(log mlog.Log) {
1153 n, err := bstore.QueryDB[MsgRetired](mox.Shutdown, DB).FilterLess("KeepUntil", time.Now()).Delete()
1154 log.Check(err, "removing old retired messages")
1156 log.Debug("cleaned up retired messages", slog.Int("count", n))
1160func startQueue(resolver dns.Resolver, done chan struct{}) {
1162 log := mlog.New("queue", nil)
1164 // Map keys are either dns.Domain.Name()'s, or string-formatted IP addresses.
1165 busyDomains := map[string]struct{}{}
1167 timer := time.NewTimer(0)
1171 case <-mox.Shutdown.Done():
1172 for len(busyDomains) > 0 {
1173 domain := <-deliveryResults
1174 delete(busyDomains, domain)
1180 case domain := <-deliveryResults:
1181 delete(busyDomains, domain)
1184 if len(busyDomains) >= maxConcurrentDeliveries {
1188 launchWork(log, resolver, busyDomains)
1189 timer.Reset(nextWork(mox.Shutdown, log, busyDomains))
1193func nextWork(ctx context.Context, log mlog.Log, busyDomains map[string]struct{}) time.Duration {
1194 q := bstore.QueryDB[Msg](ctx, DB)
1195 if len(busyDomains) > 0 {
1197 for d := range busyDomains {
1198 doms = append(doms, d)
1200 q.FilterNotEqual("RecipientDomainStr", doms...)
1202 q.FilterEqual("Hold", false)
1203 q.SortAsc("NextAttempt")
1206 if err == bstore.ErrAbsent {
1207 return 24 * time.Hour
1208 } else if err != nil {
1209 log.Errorx("finding time for next delivery attempt", err)
1210 return 1 * time.Minute
1212 return time.Until(qm.NextAttempt)
1215func launchWork(log mlog.Log, resolver dns.Resolver, busyDomains map[string]struct{}) int {
1216 q := bstore.QueryDB[Msg](mox.Shutdown, DB)
1217 q.FilterLessEqual("NextAttempt", time.Now())
1218 q.FilterEqual("Hold", false)
1219 q.SortAsc("NextAttempt")
1220 q.Limit(maxConcurrentDeliveries)
1221 if len(busyDomains) > 0 {
1223 for d := range busyDomains {
1224 doms = append(doms, d)
1226 q.FilterNotEqual("RecipientDomainStr", doms...)
1229 seen := map[string]bool{}
1230 err := q.ForEach(func(m Msg) error {
1231 dom := m.RecipientDomainStr
1232 if _, ok := busyDomains[dom]; !ok && !seen[dom] {
1234 msgs = append(msgs, m)
1239 log.Errorx("querying for work in queue", err)
1240 mox.Sleep(mox.Shutdown, 1*time.Second)
1244 for _, m := range msgs {
1245 busyDomains[m.RecipientDomainStr] = struct{}{}
1246 go deliver(log, resolver, m)
1251// todo future: we may consider keeping message files around for a while after retiring. especially for failures to deliver. to inspect what exactly wasn't delivered.
1253func removeMsgsFS(log mlog.Log, msgs ...Msg) error {
1255 for _, m := range msgs {
1256 p := mox.DataDirPath(filepath.Join("queue", store.MessagePath(m.ID)))
1257 if err := os.Remove(p); err != nil {
1258 errs = append(errs, fmt.Sprintf("%s: %v", p, err))
1262 return fmt.Errorf("removing message files from queue: %s", strings.Join(errs, "; "))
1267// Move one or more messages to retire list or remove it. Webhooks are scheduled.
1268// IDs of msgs in suppressedMsgIDs caused a suppression to be added.
1270// Callers should update Msg.Results before calling.
1272// Callers must remove the messages from the file system afterwards, see
1273// removeMsgsFS. Callers must also kick the message and webhook queues.
1274func retireMsgs(log mlog.Log, tx *bstore.Tx, event webhook.OutgoingEvent, code int, secode string, suppressedMsgIDs []int64, msgs ...Msg) error {
1279 accConf, ok := mox.Conf.Account(m0.SenderAccount)
1281 if accConf.OutgoingWebhook != nil {
1282 hookURL = accConf.OutgoingWebhook.URL
1284 log.Debug("retiring messages from queue", slog.Any("event", event), slog.String("account", m0.SenderAccount), slog.Bool("ok", ok), slog.String("webhookurl", hookURL))
1285 if hookURL != "" && (len(accConf.OutgoingWebhook.Events) == 0 || slices.Contains(accConf.OutgoingWebhook.Events, string(event))) {
1286 for _, m := range msgs {
1287 suppressing := slices.Contains(suppressedMsgIDs, m.ID)
1288 h, err := hookCompose(m, hookURL, accConf.OutgoingWebhook.Authorization, event, suppressing, code, secode)
1290 log.Errorx("composing webhooks while retiring messages from queue, not queueing hook for message", err, slog.Int64("msgid", m.ID), slog.Any("recipient", m.Recipient()))
1292 hooks = append(hooks, h)
1297 msgKeep := 24 * 7 * time.Hour
1298 hookKeep := 24 * 7 * time.Hour
1300 msgKeep = accConf.KeepRetiredMessagePeriod
1301 hookKeep = accConf.KeepRetiredWebhookPeriod
1304 for _, m := range msgs {
1305 if err := tx.Delete(&m); err != nil {
1310 for _, m := range msgs {
1311 rm := m.Retired(event == webhook.EventDelivered, now, now.Add(msgKeep))
1312 if err := tx.Insert(&rm); err != nil {
1318 for i := range hooks {
1319 if err := hookInsert(tx, &hooks[i], now, hookKeep); err != nil {
1320 return fmt.Errorf("enqueueing webhooks while retiring messages from queue: %v", err)
1325 for _, h := range hooks {
1326 log.Debug("queued webhook while retiring message from queue", h.attrs()...)
1333// deliver attempts to deliver a message.
1334// The queue is updated, either by removing a delivered or permanently failed
1335// message, or updating the time for the next attempt. A DSN may be sent.
1336func deliver(log mlog.Log, resolver dns.Resolver, m0 Msg) {
1339 qlog := log.WithCid(mox.Cid()).With(
1340 slog.Any("from", m0.Sender()),
1341 slog.Int("attempts", m0.Attempts))
1344 deliveryResults <- formatIPDomain(m0.RecipientDomain)
1348 qlog.Error("deliver panic", slog.Any("panic", x), slog.Int64("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))
1350 metrics.PanicInc(metrics.Queue)
1354 // We'll use a single transaction for the various checks, committing as soon as
1355 // we're done with it.
1356 xtx, err := DB.Begin(mox.Shutdown, true)
1358 qlog.Errorx("transaction for gathering messages to deliver", err)
1363 err := xtx.Rollback()
1364 qlog.Check(err, "rolling back transaction after error delivering")
1368 // We register this attempt by setting LastAttempt, adding an empty Result, and
1369 // already setting NextAttempt in the future with exponential backoff. If we run
1370 // into trouble delivery below, at least we won't be bothering the receiving server
1371 // with our problems.
1372 // Delivery attempts: immediately, 7.5m, 15m, 30m, 1h, 2h (send delayed DSN), 4h,
1373 // 8h, 16h (send permanent failure DSN).
1377 var backoff time.Duration
1378 var origNextAttempt time.Time
1379 prepare := func() error {
1380 // Refresh message within transaction.
1382 if err := xtx.Get(&m0); err != nil {
1383 return fmt.Errorf("get message to be delivered: %v", err)
1386 backoff = time.Duration(7*60+30+jitter.IntN(10)-5) * time.Second
1387 for range m0.Attempts {
1388 backoff *= time.Duration(2)
1391 origNextAttempt = m0.NextAttempt
1392 m0.LastAttempt = &now
1393 m0.NextAttempt = now.Add(backoff)
1394 m0.Results = append(m0.Results, MsgResult{Start: now, Error: resultErrorDelivering})
1395 if err := xtx.Update(&m0); err != nil {
1396 return fmt.Errorf("update message to be delivered: %v", err)
1400 if err := prepare(); err != nil {
1401 qlog.Errorx("storing delivery attempt", err, slog.Int64("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))
1407 // If domain of sender is currently disabled, fail the delivery attempt.
1408 if domConf, _ := mox.Conf.Domain(m0.SenderDomain.Domain); domConf.Disabled {
1409 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, fmt.Errorf("domain of sender temporarily disabled"))
1411 qlog.Check(err, "commit processing failure to deliver messages")
1417 // Check if recipient is on suppression list. If so, fail delivery.
1418 path := smtp.Path{Localpart: m0.RecipientLocalpart, IPDomain: m0.RecipientDomain}
1419 baseAddr := baseAddress(path).XString(true)
1420 qsup := bstore.QueryTx[webapi.Suppression](xtx)
1421 qsup.FilterNonzero(webapi.Suppression{Account: m0.SenderAccount, BaseAddress: baseAddr})
1422 exists, err := qsup.Exists()
1423 if err != nil || exists {
1425 qlog.Errorx("checking whether recipient address is in suppression list", err)
1427 err := fmt.Errorf("not delivering to recipient address %s: %w", path.XString(true), errSuppressed)
1428 err = smtpclient.Error{Permanent: true, Err: err}
1429 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, err)
1432 qlog.Check(err, "commit processing failure to deliver messages")
1438 resolveTransport := func(mm Msg) (string, config.Transport, bool) {
1439 if mm.Transport != "" {
1440 transport, ok := mox.Conf.Static.Transports[mm.Transport]
1442 return "", config.Transport{}, false
1444 return mm.Transport, transport, ok
1446 route := findRoute(mm.Attempts, mm)
1447 return route.Transport, route.ResolvedTransport, true
1450 // Find route for transport to use for delivery attempt.
1452 transportName, transport, transportOK := resolveTransport(m0)
1455 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, fmt.Errorf("cannot find transport %q", m0.Transport))
1457 qlog.Check(err, "commit processing failure to deliver messages")
1463 if transportName != "" {
1464 qlog = qlog.With(slog.String("transport", transportName))
1465 qlog.Debug("delivering with transport")
1468 // Attempt to gather more recipients for this identical message, only with the same
1469 // recipient domain, and under the same conditions (recipientdomain, attempts,
1473 gather := func() error {
1474 q := bstore.QueryTx[Msg](xtx)
1475 q.FilterNonzero(Msg{BaseID: m0.BaseID, RecipientDomainStr: m0.RecipientDomainStr, Attempts: m0.Attempts - 1})
1476 q.FilterNotEqual("ID", m0.ID)
1477 q.FilterLessEqual("NextAttempt", origNextAttempt)
1478 q.FilterEqual("Hold", false)
1479 err := q.ForEach(func(xm Msg) error {
1480 mrtls := m0.RequireTLS != nil
1481 xmrtls := xm.RequireTLS != nil
1482 if mrtls != xmrtls || mrtls && *m0.RequireTLS != *xm.RequireTLS {
1485 tn, _, ok := resolveTransport(xm)
1486 if ok && tn == transportName {
1487 msgs = append(msgs, &xm)
1492 return fmt.Errorf("looking up more recipients: %v", err)
1495 // Mark these additional messages as attempted too.
1496 for _, mm := range msgs[1:] {
1498 mm.NextAttempt = m0.NextAttempt
1499 mm.LastAttempt = m0.LastAttempt
1500 mm.Results = append(mm.Results, MsgResult{Start: now, Error: resultErrorDelivering})
1501 if err := xtx.Update(mm); err != nil {
1502 return fmt.Errorf("updating more message recipients for smtp transaction: %v", err)
1507 if err := gather(); err != nil {
1508 qlog.Errorx("error finding more recipients for message, will attempt to send to single recipient", err)
1513 if err := xtx.Commit(); err != nil {
1514 qlog.Errorx("commit of preparation to deliver", err, slog.Any("msgid", m0.ID))
1520 ids := make([]int64, len(msgs))
1521 rcpts := make([]smtp.Path, len(msgs))
1522 for i, m := range msgs {
1524 rcpts[i] = m.Recipient()
1526 qlog.Debug("delivering to multiple recipients", slog.Any("msgids", ids), slog.Any("recipients", rcpts))
1528 qlog.Debug("delivering to single recipient", slog.Any("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))
1532 deliverLocalserve(ctx, qlog, msgs, backoff)
1536 // We gather TLS connection successes and failures during delivery, and we store
1537 // them in tlsrptdb. Every 24 hours we send an email with a report to the recipient
1538 // domains that opt in via a TLSRPT DNS record. For us, the tricky part is
1539 // collecting all reporting information. We've got several TLS modes
1540 // (opportunistic, DANE and/or MTA-STS (PKIX), overrides due to Require TLS).
1541 // Failures can happen at various levels: MTA-STS policies (apply to whole delivery
1542 // attempt/domain), MX targets (possibly multiple per delivery attempt, both for
1543 // MTA-STS and DANE).
1545 // Once the SMTP client has tried a TLS handshake, we register success/failure,
1546 // regardless of what happens next on the connection. We also register failures
1547 // when they happen before we get to the SMTP client, but only if they are related
1548 // to TLS (and some DNSSEC).
1549 var recipientDomainResult tlsrpt.Result
1550 var hostResults []tlsrpt.Result
1552 if mox.Conf.Static.NoOutgoingTLSReports || m0.RecipientDomain.IsIP() {
1557 dayUTC := now.UTC().Format("20060102")
1559 // See if this contains a failure. If not, we'll mark TLS results for delivering
1560 // DMARC reports SendReport false, so we won't as easily get into a report sending
1563 for _, result := range hostResults {
1564 if result.Summary.TotalFailureSessionCount > 0 {
1569 if recipientDomainResult.Summary.TotalFailureSessionCount > 0 {
1573 results := make([]tlsrptdb.TLSResult, 0, 1+len(hostResults))
1574 tlsaPolicyDomains := map[string]bool{}
1575 addResult := func(r tlsrpt.Result, isHost bool) {
1576 var zerotype tlsrpt.PolicyType
1577 if r.Policy.Type == zerotype {
1581 // Ensure we store policy domain in unicode in database.
1582 policyDomain, err := dns.ParseDomain(r.Policy.Domain)
1584 qlog.Errorx("parsing policy domain for tls result", err, slog.String("policydomain", r.Policy.Domain))
1588 if r.Policy.Type == tlsrpt.TLSA {
1589 tlsaPolicyDomains[policyDomain.ASCII] = true
1592 tlsResult := tlsrptdb.TLSResult{
1593 PolicyDomain: policyDomain.Name(),
1595 RecipientDomain: m0.RecipientDomain.Domain.Name(),
1597 SendReport: !m0.IsTLSReport && (!m0.IsDMARCReport || failure),
1598 Results: []tlsrpt.Result{r},
1600 results = append(results, tlsResult)
1602 for _, result := range hostResults {
1603 addResult(result, true)
1605 // If we were delivering to a mail host directly (not a domain with MX records), we
1606 // are more likely to get a TLSA policy than an STS policy. Don't potentially
1607 // confuse operators with both a tlsa and no-policy-found result.
1609 if recipientDomainResult.Policy.Type != tlsrpt.NoPolicyFound || !tlsaPolicyDomains[recipientDomainResult.Policy.Domain] {
1610 addResult(recipientDomainResult, false)
1613 if len(results) > 0 {
1614 err := tlsrptdb.AddTLSResults(context.Background(), results)
1615 qlog.Check(err, "adding tls results to database for upcoming tlsrpt report")
1619 var dialer smtpclient.Dialer = &net.Dialer{}
1620 if transport.Submissions != nil {
1621 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submissions, true, 465)
1622 } else if transport.Submission != nil {
1623 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submission, false, 587)
1624 } else if transport.SMTP != nil {
1625 // todo future: perhaps also gather tlsrpt results for submissions.
1626 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.SMTP, false, 25)
1628 ourHostname := mox.Conf.Static.HostnameDomain
1629 if transport.Socks != nil {
1630 socksdialer, err := proxy.SOCKS5("tcp", transport.Socks.Address, nil, &net.Dialer{})
1632 failMsgsDB(qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer: %v", err))
1634 } else if d, ok := socksdialer.(smtpclient.Dialer); !ok {
1635 failMsgsDB(qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer is not a contextdialer"))
1640 ourHostname = transport.Socks.Hostname
1642 recipientDomainResult, hostResults = deliverDirect(qlog, resolver, dialer, ourHostname, transportName, transport.Direct, msgs, backoff)
1646func findRoute(attempt int, m Msg) config.Route {
1647 routesAccount, routesDomain, routesGlobal := mox.Conf.Routes(m.SenderAccount, m.SenderDomain.Domain)
1648 if r, ok := findRouteInList(attempt, m, routesAccount); ok {
1651 if r, ok := findRouteInList(attempt, m, routesDomain); ok {
1654 if r, ok := findRouteInList(attempt, m, routesGlobal); ok {
1657 return config.Route{}
1660func findRouteInList(attempt int, m Msg, routes []config.Route) (config.Route, bool) {
1661 for _, r := range routes {
1662 if routeMatch(attempt, m, r) {
1666 return config.Route{}, false
1669func routeMatch(attempt int, m Msg, r config.Route) bool {
1670 return attempt >= r.MinimumAttempts && routeMatchDomain(r.FromDomainASCII, m.SenderDomain.Domain) && routeMatchDomain(r.ToDomainASCII, m.RecipientDomain.Domain)
1673func routeMatchDomain(l []string, d dns.Domain) bool {
1677 for _, e := range l {
1678 if d.ASCII == e || strings.HasPrefix(e, ".") && (d.ASCII == e[1:] || strings.HasSuffix(d.ASCII, e)) {
1685// Returns string representing delivery result for err, and number of delivered and
1688// Values: ok, okpartial, timeout, canceled, temperror, permerror, error.
1689func deliveryResult(err error, delivered, failed int) string {
1690 var cerr smtpclient.Error
1695 } else if failed > 0 {
1699 case errors.Is(err, os.ErrDeadlineExceeded), errors.Is(err, context.DeadlineExceeded):
1701 case errors.Is(err, context.Canceled):
1703 case errors.As(err, &cerr):