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.

4package queue

5

6import (

7 "context"

8 "fmt"

9 "io"

10 "net"

11 "os"

12 "path/filepath"

13 "runtime/debug"

14 "sort"

15 "strings"

16 "time"

17

18 "golang.org/x/net/proxy"

19

20 "github.com/prometheus/client_golang/prometheus"

21 "github.com/prometheus/client_golang/prometheus/promauto"

22

23 "github.com/mjl-/bstore"

24

25 "github.com/mjl-/mox/config"

26 "github.com/mjl-/mox/dns"

27 "github.com/mjl-/mox/dsn"

28 "github.com/mjl-/mox/metrics"

29 "github.com/mjl-/mox/mlog"

30 "github.com/mjl-/mox/mox-"

31 "github.com/mjl-/mox/moxio"

32 "github.com/mjl-/mox/smtp"

33 "github.com/mjl-/mox/store"

34)

35

36var xlog = mlog.New("queue")

37

38var (

39 metricConnection = promauto.NewCounterVec(

40 prometheus.CounterOpts{

41 Name: "mox_queue_connection_total",

42 Help: "Queue client connections, outgoing.",

43 },

44 []string{

45 "result", // "ok", "timeout", "canceled", "error"

46 },

47 )

48 metricDelivery = promauto.NewHistogramVec(

49 prometheus.HistogramOpts{

50 Name: "mox_queue_delivery_duration_seconds",

51 Help: "SMTP client delivery attempt to single host.",

52 Buckets: []float64{0.01, 0.05, 0.100, 0.5, 1, 5, 10, 20, 30, 60, 120},

53 },

54 []string{

55 "attempt", // Number of attempts.

56 "transport", // empty for default direct delivery.

57 "tlsmode", // strict, opportunistic, skip

58 "result", // ok, timeout, canceled, temperror, permerror, error

59 },

60 )

61)

62

63type contextDialer interface {

64 DialContext(ctx context.Context, network, addr string) (c net.Conn, err error)

65}

66

67// Used to dial remote SMTP servers.

68// Overridden for tests.

69var dial = func(ctx context.Context, dialer contextDialer, timeout time.Duration, addr string, laddr net.Addr) (net.Conn, error) {

70 // If this is a net.Dialer, use its settings and add the timeout and localaddr.

71 // This is the typical case, but SOCKS5 support can use a different dialer.

72 if d, ok := dialer.(*net.Dialer); ok {

73 nd := *d

74 nd.Timeout = timeout

75 nd.LocalAddr = laddr

76 return nd.DialContext(ctx, "tcp", addr)

77 }

78 return dialer.DialContext(ctx, "tcp", addr)

79}

80

81var jitter = mox.NewRand()

82

83var DBTypes = []any{Msg{}} // Types stored in DB.

84var DB *bstore.DB // Exported for making backups.

85

86// Set for mox localserve, to prevent queueing.

87var Localserve bool

88

89// Msg is a message in the queue.

90type Msg struct {

91 ID int64

92 Queued time.Time `bstore:"default now"`

93 SenderAccount string // Failures are delivered back to this local account. Also used for routing.

94 SenderLocalpart smtp.Localpart // Should be a local user and domain.

95 SenderDomain dns.IPDomain

96 RecipientLocalpart smtp.Localpart // Typically a remote user and domain.

97 RecipientDomain dns.IPDomain

98 RecipientDomainStr string // For filtering.

99 Attempts int // Next attempt is based on last attempt and exponential back off based on attempts.

100 DialedIPs map[string][]net.IP // For each host, the IPs that were dialed. Used for IP selection for later attempts.

101 NextAttempt time.Time // For scheduling.

102 LastAttempt *time.Time

103 LastError string

104 Has8bit bool // Whether message contains bytes with high bit set, determines whether 8BITMIME SMTP extension is needed.

105 SMTPUTF8 bool // Whether message requires use of SMTPUTF8.

106 Size int64 // Full size of message, combined MsgPrefix with contents of message file.

107 MessageID string // Used when composing a DSN, in its References header.

108 MsgPrefix []byte

109

110 // If set, this message is a DSN and this is a version using utf-8, for the case

111 // the remote MTA supports smtputf8. In this case, Size and MsgPrefix are not

112 // relevant.

113 DSNUTF8 []byte

114

115 // If non-empty, the transport to use for this message. Can be set through cli or

116 // admin interface. If empty (the default for a submitted message), regular routing

117 // rules apply.

118 Transport string

119}

120

121// Sender of message as used in MAIL FROM.

122func (m Msg) Sender() smtp.Path {

123 return smtp.Path{Localpart: m.SenderLocalpart, IPDomain: m.SenderDomain}

124}

125

126// Recipient of message as used in RCPT TO.

127func (m Msg) Recipient() smtp.Path {

128 return smtp.Path{Localpart: m.RecipientLocalpart, IPDomain: m.RecipientDomain}

129}

130

131// MessagePath returns the path where the message is stored.

132func (m Msg) MessagePath() string {

133 return mox.DataDirPath(filepath.Join("queue", store.MessagePath(m.ID)))

134}

135

136// Init opens the queue database without starting delivery.

137func Init() error {

138 qpath := mox.DataDirPath("queue/index.db")

139 os.MkdirAll(filepath.Dir(qpath), 0770)

140 isNew := false

141 if _, err := os.Stat(qpath); err != nil && os.IsNotExist(err) {

142 isNew = true

143 }

144

145 var err error

146 DB, err = bstore.Open(mox.Shutdown, qpath, &bstore.Options{Timeout: 5 * time.Second, Perm: 0660}, DBTypes...)

147 if err != nil {

148 if isNew {

149 os.Remove(qpath)

150 }

151 return fmt.Errorf("open queue database: %s", err)

152 }

153 return nil

154}

155

156// Shutdown closes the queue database. The delivery process isn't stopped. For tests only.

157func Shutdown() {

158 err := DB.Close()

159 xlog.Check(err, "closing queue db")

160 DB = nil

161}

162

163// List returns all messages in the delivery queue.

164// Ordered by earliest delivery attempt first.

165func List(ctx context.Context) ([]Msg, error) {

166 qmsgs, err := bstore.QueryDB[Msg](ctx, DB).List()

167 if err != nil {

168 return nil, err

169 }

170 sort.Slice(qmsgs, func(i, j int) bool {

171 a := qmsgs[i]

172 b := qmsgs[j]

173 la := a.LastAttempt != nil

174 lb := b.LastAttempt != nil

175 if !la && lb {

176 return true

177 } else if la && !lb {

178 return false

179 }

180 if !la && !lb || a.LastAttempt.Equal(*b.LastAttempt) {

181 return a.ID < b.ID

182 }

183 return a.LastAttempt.Before(*b.LastAttempt)

184 })

185 return qmsgs, nil

186}

187

188// Count returns the number of messages in the delivery queue.

189func Count(ctx context.Context) (int, error) {

190 return bstore.QueryDB[Msg](ctx, DB).Count()

191}

192

193// Add a new message to the queue. The queue is kicked immediately to start a

194// first delivery attempt.

195//

196// If consumeFile is true, it is removed as part of delivery (by rename or copy

197// and remove). msgFile is never closed by Add.

198//

199// dnsutf8Opt is a utf8-version of the message, to be used only for DNSs. If set,

200// this data is used as the message when delivering the DSN and the remote SMTP

201// server supports SMTPUTF8. If the remote SMTP server does not support SMTPUTF8,

202// the regular non-utf8 message is delivered.

203func Add(ctx context.Context, log *mlog.Log, senderAccount string, mailFrom, rcptTo smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, msgPrefix []byte, msgFile *os.File, dsnutf8Opt []byte, consumeFile bool) (int64, error) {

204 // todo: Add should accept multiple rcptTo if they are for the same domain. so we can queue them for delivery in one (or just a few) session(s), transferring the data only once. ../rfc/5321:3759

205

206 if Localserve {

207 if senderAccount == "" {

208 return 0, fmt.Errorf("cannot queue with localserve without local account")

209 }

210 acc, err := store.OpenAccount(senderAccount)

211 if err != nil {

212 return 0, fmt.Errorf("opening sender account for immediate delivery with localserve: %v", err)

213 }

214 defer func() {

215 err := acc.Close()

216 log.Check(err, "closing account")

217 }()

218 m := store.Message{Size: size, MsgPrefix: msgPrefix}

219 conf, _ := acc.Conf()

220 dest := conf.Destinations[mailFrom.String()]

221 acc.WithWLock(func() {

222 err = acc.DeliverDestination(log, dest, &m, msgFile, consumeFile)

223 })

224 if err != nil {

225 return 0, fmt.Errorf("delivering message: %v", err)

226 }

227 log.Debug("immediately delivered from queue to sender")

228 return 0, nil

229 }

230

231 tx, err := DB.Begin(ctx, true)

232 if err != nil {

233 return 0, fmt.Errorf("begin transaction: %w", err)

234 }

235 defer func() {

236 if tx != nil {

237 if err := tx.Rollback(); err != nil {

238 log.Errorx("rollback for queue", err)

239 }

240 }

241 }()

242

243 now := time.Now()

244 qm := Msg{0, now, senderAccount, mailFrom.Localpart, mailFrom.IPDomain, rcptTo.Localpart, rcptTo.IPDomain, formatIPDomain(rcptTo.IPDomain), 0, nil, now, nil, "", has8bit, smtputf8, size, messageID, msgPrefix, dsnutf8Opt, ""}

245

246 if err := tx.Insert(&qm); err != nil {

247 return 0, err

248 }

249

250 dst := qm.MessagePath()

251 defer func() {

252 if dst != "" {

253 err := os.Remove(dst)

254 log.Check(err, "removing destination message file for queue", mlog.Field("path", dst))

255 }

256 }()

257 dstDir := filepath.Dir(dst)

258 os.MkdirAll(dstDir, 0770)

259 if consumeFile {

260 if err := os.Rename(msgFile.Name(), dst); err != nil {

261 // Could be due to cross-filesystem rename. Users shouldn't configure their systems that way.

262 return 0, fmt.Errorf("move message into queue dir: %w", err)

263 }

264 } else if err := moxio.LinkOrCopy(log, dst, msgFile.Name(), nil, true); err != nil {

265 return 0, fmt.Errorf("linking/copying message to new file: %s", err)

266 } else if err := moxio.SyncDir(dstDir); err != nil {

267 return 0, fmt.Errorf("sync directory: %v", err)

268 }

269

270 if err := tx.Commit(); err != nil {

271 return 0, fmt.Errorf("commit transaction: %s", err)

272 }

273 tx = nil

274 dst = ""

275

276 queuekick()

277 return qm.ID, nil

278}

279

280func formatIPDomain(d dns.IPDomain) string {

281 if len(d.IP) > 0 {

282 return "[" + d.IP.String() + "]"

283 }

284 return d.Domain.Name()

285}

286

287var (

288 kick = make(chan struct{}, 1)

289 deliveryResult = make(chan string, 1)

290)

291

292func queuekick() {

293 select {

294 case kick <- struct{}{}:

295 default:

296 }

297}

298

299// Kick sets the NextAttempt for messages matching all filter parameters (ID,

300// toDomain, recipient) that are nonzero, and kicks the queue, attempting delivery

301// of those messages. If all parameters are zero, all messages are kicked. If

302// transport is set, the delivery attempts for the matching messages will use the

303// transport. An empty string is the default transport, i.e. direct delivery.

304// Returns number of messages queued for immediate delivery.

305func Kick(ctx context.Context, ID int64, toDomain, recipient string, transport *string) (int, error) {

306 q := bstore.QueryDB[Msg](ctx, DB)

307 if ID > 0 {

308 q.FilterID(ID)

309 }

310 if toDomain != "" {

311 q.FilterEqual("RecipientDomainStr", toDomain)

312 }

313 if recipient != "" {

314 q.FilterFn(func(qm Msg) bool {

315 return qm.Recipient().XString(true) == recipient

316 })

317 }

318 up := map[string]any{"NextAttempt": time.Now()}

319 if transport != nil {

320 if *transport != "" {

321 _, ok := mox.Conf.Static.Transports[*transport]

322 if !ok {

323 return 0, fmt.Errorf("unknown transport %q", *transport)

324 }

325 }

326 up["Transport"] = *transport

327 }

328 n, err := q.UpdateFields(up)

329 if err != nil {

330 return 0, fmt.Errorf("selecting and updating messages in queue: %v", err)

331 }

332 queuekick()

333 return n, nil

334}

335

336// Drop removes messages from the queue that match all nonzero parameters.

337// If all parameters are zero, all messages are removed.

338// Returns number of messages removed.

339func Drop(ctx context.Context, ID int64, toDomain string, recipient string) (int, error) {

340 q := bstore.QueryDB[Msg](ctx, DB)

341 if ID > 0 {

342 q.FilterID(ID)

343 }

344 if toDomain != "" {

345 q.FilterEqual("RecipientDomainStr", toDomain)

346 }

347 if recipient != "" {

348 q.FilterFn(func(qm Msg) bool {

349 return qm.Recipient().XString(true) == recipient

350 })

351 }

352 var msgs []Msg

353 q.Gather(&msgs)

354 n, err := q.Delete()

355 if err != nil {

356 return 0, fmt.Errorf("selecting and deleting messages from queue: %v", err)

357 }

358 for _, m := range msgs {

359 p := m.MessagePath()

360 if err := os.Remove(p); err != nil {

361 xlog.WithContext(ctx).Errorx("removing queue message from file system", err, mlog.Field("queuemsgid", m.ID), mlog.Field("path", p))

362 }

363 }

364 return n, nil

365}

366

367type ReadReaderAtCloser interface {

368 io.ReadCloser

369 io.ReaderAt

370}

371

372// OpenMessage opens a message present in the queue.

373func OpenMessage(ctx context.Context, id int64) (ReadReaderAtCloser, error) {

374 qm := Msg{ID: id}

375 err := DB.Get(ctx, &qm)

376 if err != nil {

377 return nil, err

378 }

379 f, err := os.Open(qm.MessagePath())

380 if err != nil {

381 return nil, fmt.Errorf("open message file: %s", err)

382 }

383 r := store.FileMsgReader(qm.MsgPrefix, f)

384 return r, err

385}

386

387const maxConcurrentDeliveries = 10

388

389// Start opens the database by calling Init, then starts the delivery process.

390func Start(resolver dns.Resolver, done chan struct{}) error {

391 if err := Init(); err != nil {

392 return err

393 }

394

395 // High-level delivery strategy advice: ../rfc/5321:3685

396 go func() {

397 // Map keys are either dns.Domain.Name()'s, or string-formatted IP addresses.

398 busyDomains := map[string]struct{}{}

399

400 timer := time.NewTimer(0)

401

402 for {

403 select {

404 case <-mox.Shutdown.Done():

405 done <- struct{}{}

406 return

407 case <-kick:

408 case <-timer.C:

409 case domain := <-deliveryResult:

410 delete(busyDomains, domain)

411 }

412

413 if len(busyDomains) >= maxConcurrentDeliveries {

414 continue

415 }

416

417 launchWork(resolver, busyDomains)

418 timer.Reset(nextWork(mox.Shutdown, busyDomains))

419 }

420 }()

421 return nil

422}

423

424func nextWork(ctx context.Context, busyDomains map[string]struct{}) time.Duration {

425 q := bstore.QueryDB[Msg](ctx, DB)

426 if len(busyDomains) > 0 {

427 var doms []any

428 for d := range busyDomains {

429 doms = append(doms, d)

430 }

431 q.FilterNotEqual("RecipientDomainStr", doms...)

432 }

433 q.SortAsc("NextAttempt")

434 q.Limit(1)

435 qm, err := q.Get()

436 if err == bstore.ErrAbsent {

437 return 24 * time.Hour

438 } else if err != nil {

439 xlog.Errorx("finding time for next delivery attempt", err)

440 return 1 * time.Minute

441 }

442 return time.Until(qm.NextAttempt)

443}

444

445func launchWork(resolver dns.Resolver, busyDomains map[string]struct{}) int {

446 q := bstore.QueryDB[Msg](mox.Shutdown, DB)

447 q.FilterLessEqual("NextAttempt", time.Now())

448 q.SortAsc("NextAttempt")

449 q.Limit(maxConcurrentDeliveries)

450 if len(busyDomains) > 0 {

451 var doms []any

452 for d := range busyDomains {

453 doms = append(doms, d)

454 }

455 q.FilterNotEqual("RecipientDomainStr", doms...)

456 }

457 msgs, err := q.List()

458 if err != nil {

459 xlog.Errorx("querying for work in queue", err)

460 mox.Sleep(mox.Shutdown, 1*time.Second)

461 return -1

462 }

463

464 for _, m := range msgs {

465 busyDomains[formatIPDomain(m.RecipientDomain)] = struct{}{}

466 go deliver(resolver, m)

467 }

468 return len(msgs)

469}

470

471// Remove message from queue in database and file system.

472func queueDelete(ctx context.Context, msgID int64) error {

473 if err := DB.Delete(ctx, &Msg{ID: msgID}); err != nil {

474 return err

475 }

476 // If removing from database fails, we'll also leave the file in the file system.

477

478 p := mox.DataDirPath(filepath.Join("queue", store.MessagePath(msgID)))

479 if err := os.Remove(p); err != nil {

480 return fmt.Errorf("removing queue message from file system: %v", err)

481 }

482

483 return nil

484}

485

486// deliver attempts to deliver a message.

487// The queue is updated, either by removing a delivered or permanently failed

488// message, or updating the time for the next attempt. A DSN may be sent.

489func deliver(resolver dns.Resolver, m Msg) {

490 cid := mox.Cid()

491 qlog := xlog.WithCid(cid).Fields(mlog.Field("from", m.Sender()), mlog.Field("recipient", m.Recipient()), mlog.Field("attempts", m.Attempts), mlog.Field("msgid", m.ID))

492

493 defer func() {

494 deliveryResult <- formatIPDomain(m.RecipientDomain)

495

496 x := recover()

497 if x != nil {

498 qlog.Error("deliver panic", mlog.Field("panic", x))

499 debug.PrintStack()

500 metrics.PanicInc(metrics.Queue)

501 }

502 }()

503

504 // We register this attempt by setting last_attempt, and already next_attempt time

505 // in the future with exponential backoff. If we run into trouble delivery below,

506 // at least we won't be bothering the receiving server with our problems.

507 // Delivery attempts: immediately, 7.5m, 15m, 30m, 1h, 2h (send delayed DSN), 4h,

508 // 8h, 16h (send permanent failure DSN).

509 // ../rfc/5321:3703

510 // todo future: make the back off times configurable. ../rfc/5321:3713

511 backoff := time.Duration(7*60+30+jitter.Intn(10)-5) * time.Second

512 for i := 0; i < m.Attempts; i++ {

513 backoff *= time.Duration(2)

514 }

515 m.Attempts++

516 now := time.Now()

517 m.LastAttempt = &now

518 m.NextAttempt = now.Add(backoff)

519 qup := bstore.QueryDB[Msg](mox.Shutdown, DB)

520 qup.FilterID(m.ID)

521 update := Msg{Attempts: m.Attempts, NextAttempt: m.NextAttempt, LastAttempt: m.LastAttempt}

522 if _, err := qup.UpdateNonzero(update); err != nil {

523 qlog.Errorx("storing delivery attempt", err)

524 return

525 }

526

527 // Find route for transport to use for delivery attempt.

528 var transport config.Transport

529 var transportName string

530 if m.Transport != "" {

531 var ok bool

532 transport, ok = mox.Conf.Static.Transports[m.Transport]

533 if !ok {

534 var remoteMTA dsn.NameIP // Zero value, will not be included in DSN. ../rfc/3464:1027

535 fail(qlog, m, backoff, false, remoteMTA, "", fmt.Sprintf("cannot find transport %q", m.Transport))

536 return

537 }

538 transportName = m.Transport

539 } else {

540 route := findRoute(m.Attempts-1, m)

541 transport = route.ResolvedTransport

542 transportName = route.Transport

543 }

544

545 if transportName != "" {

546 qlog = qlog.Fields(mlog.Field("transport", transportName))

547 qlog.Debug("delivering with transport", mlog.Field("transport", transportName))

548 }

549

550 var dialer contextDialer = &net.Dialer{}

551 if transport.Submissions != nil {

552 deliverSubmit(cid, qlog, resolver, dialer, m, backoff, transportName, transport.Submissions, true, 465)

553 } else if transport.Submission != nil {

554 deliverSubmit(cid, qlog, resolver, dialer, m, backoff, transportName, transport.Submission, false, 587)

555 } else if transport.SMTP != nil {

556 deliverSubmit(cid, qlog, resolver, dialer, m, backoff, transportName, transport.SMTP, false, 25)

557 } else {

558 ourHostname := mox.Conf.Static.HostnameDomain

559 if transport.Socks != nil {

560 socksdialer, err := proxy.SOCKS5("tcp", transport.Socks.Address, nil, &net.Dialer{})

561 if err != nil {

562 fail(qlog, m, backoff, false, dsn.NameIP{}, "", fmt.Sprintf("socks dialer: %v", err))

563 return

564 } else if d, ok := socksdialer.(contextDialer); !ok {

565 fail(qlog, m, backoff, false, dsn.NameIP{}, "", "socks dialer is not a contextdialer")

566 return

567 } else {

568 dialer = d

569 }

570 ourHostname = transport.Socks.Hostname

571 }

572 deliverDirect(cid, qlog, resolver, dialer, ourHostname, transportName, m, backoff)

573 }

574}

575

576func findRoute(attempt int, m Msg) config.Route {

577 routesAccount, routesDomain, routesGlobal := mox.Conf.Routes(m.SenderAccount, m.SenderDomain.Domain)

578 if r, ok := findRouteInList(attempt, m, routesAccount); ok {

579 return r

580 }

581 if r, ok := findRouteInList(attempt, m, routesDomain); ok {

582 return r

583 }

584 if r, ok := findRouteInList(attempt, m, routesGlobal); ok {

585 return r

586 }

587 return config.Route{}

588}

589

590func findRouteInList(attempt int, m Msg, routes []config.Route) (config.Route, bool) {

591 for _, r := range routes {

592 if routeMatch(attempt, m, r) {

593 return r, true

594 }

595 }

596 return config.Route{}, false

597}

598

599func routeMatch(attempt int, m Msg, r config.Route) bool {

600 return attempt >= r.MinimumAttempts && routeMatchDomain(r.FromDomainASCII, m.SenderDomain.Domain) && routeMatchDomain(r.ToDomainASCII, m.RecipientDomain.Domain)

601}

602

603func routeMatchDomain(l []string, d dns.Domain) bool {

604 if len(l) == 0 {

605 return true

606 }

607 for _, e := range l {

608 if d.ASCII == e || strings.HasPrefix(e, ".") && (d.ASCII == e[1:] || strings.HasSuffix(d.ASCII, e)) {

609 return true

610 }

611 }

612 return false

613}

614

615// dialHost dials host for delivering Msg, taking previous attempts into accounts.

616// If the previous attempt used IPv4, this attempt will use IPv6 (in case one of the IPs is in a DNSBL).

617// The second attempt for an address family we prefer the same IP as earlier, to increase our chances if remote is doing greylisting.

618// dialHost updates m with the dialed IP and m should be saved in case of failure.

619// If we have fully specified local smtp listen IPs, we set those for the outgoing

620// connection. The admin probably configured these same IPs in SPF, but others

621// possibly not.

622func dialHost(ctx context.Context, log *mlog.Log, resolver dns.Resolver, dialer contextDialer, host dns.IPDomain, port int, m *Msg) (conn net.Conn, ip net.IP, dualstack bool, rerr error) {

623 var ips []net.IP

624 if len(host.IP) > 0 {

625 ips = []net.IP{host.IP}

626 } else {

627 // todo: The Go resolver automatically follows CNAMEs, which is not allowed for

628 // host names in MX records. ../rfc/5321:3861 ../rfc/2181:661

629 name := host.Domain.ASCII + "."

630 ipaddrs, err := resolver.LookupIPAddr(ctx, name)

631 if err != nil || len(ipaddrs) == 0 {

632 return nil, nil, false, fmt.Errorf("looking up %q: %v", name, err)

633 }

634 var have4, have6 bool

635 for _, ipaddr := range ipaddrs {

636 ips = append(ips, ipaddr.IP)

637 if ipaddr.IP.To4() == nil {

638 have6 = true

639 } else {

640 have4 = true

641 }

642 }

643 dualstack = have4 && have6

644 prevIPs := m.DialedIPs[host.String()]

645 if len(prevIPs) > 0 {

646 prevIP := prevIPs[len(prevIPs)-1]

647 prevIs4 := prevIP.To4() != nil

648 sameFamily := 0

649 for _, ip := range prevIPs {

650 is4 := ip.To4() != nil

651 if prevIs4 == is4 {

652 sameFamily++

653 }

654 }

655 preferPrev := sameFamily == 1

656 // We use stable sort so any preferred/randomized listing from DNS is kept intact.

657 sort.SliceStable(ips, func(i, j int) bool {

658 aIs4 := ips[i].To4() != nil

659 bIs4 := ips[j].To4() != nil

660 if aIs4 != bIs4 {

661 // Prefer "i" if it is not same address family.

662 return aIs4 != prevIs4

663 }

664 // Prefer "i" if it is the same as last and we should be preferring it.

665 return preferPrev && ips[i].Equal(prevIP)

666 })

667 log.Debug("ordered ips for dialing", mlog.Field("ips", ips))

668 }

669 }

670

671 var timeout time.Duration

672 deadline, ok := ctx.Deadline()

673 if !ok {

674 timeout = 30 * time.Second

675 } else {

676 timeout = time.Until(deadline) / time.Duration(len(ips))

677 }

678

679 var lastErr error

680 var lastIP net.IP

681 for _, ip := range ips {

682 addr := net.JoinHostPort(ip.String(), fmt.Sprintf("%d", port))

683 log.Debug("dialing remote host for delivery", mlog.Field("addr", addr))

684 var laddr net.Addr

685 for _, lip := range mox.Conf.Static.SpecifiedSMTPListenIPs {

686 ipIs4 := ip.To4() != nil

687 lipIs4 := lip.To4() != nil

688 if ipIs4 == lipIs4 {

689 laddr = &net.TCPAddr{IP: lip}

690 break

691 }

692 }

693 conn, err := dial(ctx, dialer, timeout, addr, laddr)

694 if err == nil {

695 log.Debug("connected for smtp delivery", mlog.Field("host", host), mlog.Field("addr", addr), mlog.Field("laddr", laddr))

696 if m.DialedIPs == nil {

697 m.DialedIPs = map[string][]net.IP{}

698 }

699 name := host.String()

700 m.DialedIPs[name] = append(m.DialedIPs[name], ip)

701 return conn, ip, dualstack, nil

702 }

703 log.Debugx("connection attempt for smtp delivery", err, mlog.Field("host", host), mlog.Field("addr", addr), mlog.Field("laddr", laddr))

704 lastErr = err

705 lastIP = ip

706 }

707 return nil, lastIP, dualstack, lastErr

708}

709