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

6import (

7 "bytes"

8 "context"

9 "errors"

10 "fmt"

11 "io"

12 "log/slog"

13 "net"

14 "os"

15 "path/filepath"

16 "runtime/debug"

17 "slices"

18 "strings"

19 "time"

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"

43)

45// ErrFromID indicate a fromid was present when adding a message to the queue, but

46// it wasn't unique.

47var ErrFromID = errors.New("fromid not unique")

49var (

50 metricConnection = promauto.NewCounterVec(

51 prometheus.CounterOpts{

52 Name: "mox_queue_connection_total",

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

54 },

55 []string{

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

57 },

58 )

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},

64 },

65 []string{

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

70 },

71 )

72 metricHold = promauto.NewGauge(

73 prometheus.GaugeOpts{

74 Name: "mox_queue_hold",

75 Help: "Messages in queue that are on hold.",

76 },

77 )

78)

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.

89var Localserve bool

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.

94type HoldRule struct {

95 ID int64

96 Account string

97 SenderDomain dns.Domain

98 RecipientDomain dns.Domain

99 SenderDomainStr string // Unicode.

100 RecipientDomainStr string // Unicode.

101}

102

103func (pr HoldRule) All() bool {

104 pr.ID = 0

105 return pr == HoldRule{}

106}

107

108func (pr HoldRule) matches(m Msg) bool {

109 return pr.All() || pr.Account == m.SenderAccount || pr.SenderDomainStr == m.SenderDomainStr || pr.RecipientDomainStr == m.RecipientDomainStr

110}

111

112// Msg is a message in the queue.

113//

114// Use MakeMsg to make a message with fields that Add needs. Add will further set

115// queueing related fields.

116type Msg struct {

117 ID int64

118

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"`

125

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

141 Results []MsgResult

142

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.

151

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

154 // relevant.

155 DSNUTF8 []byte

156

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

159 // rules apply.

160 Transport string

161

162 // RequireTLS influences TLS verification during delivery.

163 //

164 // If nil, the recipient domain policy is followed (MTA-STS and/or DANE), falling

165 // back to optional opportunistic non-verified STARTTLS.

166 //

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

169 // server.

170 //

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.

174 RequireTLS *bool

175 // ../rfc/8689:250

176

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

179 // utc date-time.

180 FutureReleaseRequest string

181 // ../rfc/4865:305

182

183 Extra map[string]string // Extra information, for transactional email.

184}

185

186// MsgResult is the result (or work in progress) of a delivery attempt.

187type MsgResult struct {

188 Start time.Time

189 Duration time.Duration

190 Success bool

191 Code int

192 Secode string

193 Error string

194 // todo: store smtp trace for failed deliveries for debugging, perhaps also for successful deliveries.

195}

196

197// Stored in MsgResult.Error while delivery is in progress. Replaced after success/error.

198const resultErrorDelivering = "delivering..."

199

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()})

204 }

205 result := &m.Results[len(m.Results)-1]

206 result.Duration = time.Since(result.Start)

207 result.Code = code

208 result.Secode = secode

209 result.Error = errmsg

210 result.Success = success

211}

212

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()}

217 }

218 return m.Results[len(m.Results)-1]

219}

220

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}

224}

225

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}

229}

230

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)))

234}

235

236// todo: store which transport (if any) was actually used in MsgResult, based on routes.

237

238// Retired returns a MsgRetired for the message, for history of deliveries.

239func (m Msg) Retired(success bool, t, keepUntil time.Time) MsgRetired {

240 return MsgRetired{

241 ID: m.ID,

242 BaseID: m.BaseID,

243 Queued: m.Queued,

244 SenderAccount: m.SenderAccount,

245 SenderLocalpart: m.SenderLocalpart,

246 SenderDomainStr: m.SenderDomainStr,

247 FromID: m.FromID,

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,

255 Results: m.Results,

256 Has8bit: m.Has8bit,

257 SMTPUTF8: m.SMTPUTF8,

258 IsDMARCReport: m.IsDMARCReport,

259 IsTLSReport: m.IsTLSReport,

260 Size: m.Size,

261 MessageID: m.MessageID,

262 Subject: m.Subject,

263 Transport: m.Transport,

264 RequireTLS: m.RequireTLS,

265 FutureReleaseRequest: m.FutureReleaseRequest,

266 Extra: m.Extra,

267

268 RecipientAddress: smtp.Path{Localpart: m.RecipientLocalpart, IPDomain: m.RecipientDomain}.XString(true),

269 Success: success,

270 LastActivity: t,

271 KeepUntil: keepUntil,

272 }

273}

274

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.

280

281 BaseID int64

282 Queued time.Time

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

294 Results []MsgResult

295

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.

303

304 Transport string

305 RequireTLS *bool

306 FutureReleaseRequest string

307

308 Extra map[string]string // Extra information, for transactional email.

309

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"`

314}

315

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)

324 }

325 } else {

326 path.IPDomain.Domain, err = dns.ParseDomain(m.SenderDomainStr)

327 }

328 return

329}

330

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}

334}

335

336// LastResult returns the last result entry, or an empty result.

337func (m MsgRetired) LastResult() MsgResult {

338 if len(m.Results) == 0 {

339 return MsgResult{}

340 }

341 return m.Results[len(m.Results)-1]

342}

343

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

345func Init() error {

346 qpath := mox.DataDirPath(filepath.FromSlash("queue/index.db"))

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

348 isNew := false

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

350 isNew = true

351 }

352

353 var err error

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...)

357 if err == nil {

358 err = DB.Read(mox.Shutdown, func(tx *bstore.Tx) error {

359 return metricHoldUpdate(tx)

360 })

361 }

362 if isNew && err == nil {

363 // Insert Msg with higher ID and remove it again. It will set the sequence for next

364 // ID so queue messages get assigned higher values (especially on new installs,

365 // among which localserve) so there can be no confusion between id's of messages in

366 // an account and id's of messages in the queue.

367 err = DB.Write(mox.Shutdown, func(tx *bstore.Tx) error {

368 m := Msg{ID: 1000 * 1000}

369 err := tx.Insert(&m)

370 if err == nil {

371 err = tx.Delete(&m)

372 }

373 return err

374 })

375 if err != nil {

376 err = fmt.Errorf("increasing Msg.ID sequence: %w", err)

377 }

378 }

379 if err != nil {

380 if isNew {

381 err := os.Remove(qpath)

382 log.Check(err, "removing new queue database file after error")

383 }

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

385 }

386 return nil

387}

388

389// When we update the gauge, we just get the full current value, not try to account

390// for adds/removes.

391func metricHoldUpdate(tx *bstore.Tx) error {

392 count, err := bstore.QueryTx[Msg](tx).FilterNonzero(Msg{Hold: true}).Count()

393 if err != nil {

394 return fmt.Errorf("querying messages on hold for metric: %v", err)

395 }

396 metricHold.Set(float64(count))

397 return nil

398}

399

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

401func Shutdown() {

402 err := DB.Close()

403 if err != nil {

404 mlog.New("queue", nil).Errorx("closing queue db", err)

405 }

406 DB = nil

407}

408

409// todo: the filtering & sorting can use improvements. too much duplicated code (variants between {Msg,Hook}{,Retired}. Sort has pagination fields, some untyped.

410

411// Filter filters messages to list or operate on. Used by admin web interface

412// and cli.

413//

414// Only non-empty/non-zero values are applied to the filter. Leaving all fields

415// empty/zero matches all messages.

416type Filter struct {

417 Max int

418 IDs []int64

419 Account string

420 From string

421 To string

422 Hold *bool

423 Submitted string // Whether submitted before/after a time relative to now. ">$duration" or "<$duration", also with "now" for duration.

424 NextAttempt string // ">$duration" or "<$duration", also with "now" for duration.

425 Transport *string

426}

427

428func (f Filter) apply(q *bstore.Query[Msg]) error {

429 if len(f.IDs) > 0 {

430 q.FilterIDs(f.IDs)

431 }

432 applyTime := func(field string, s string) error {

433 orig := s

434 var before bool

435 if strings.HasPrefix(s, "<") {

436 before = true

437 } else if !strings.HasPrefix(s, ">") {

438 return fmt.Errorf(`must start with "<" for before or ">" for after a duration`)

439 }

440 s = strings.TrimSpace(s[1:])

441 var t time.Time

442 if s == "now" {

443 t = time.Now()

444 } else if d, err := time.ParseDuration(s); err != nil {

445 return fmt.Errorf("parsing duration %q: %v", orig, err)

446 } else {

447 t = time.Now().Add(d)

448 }

449 if before {

450 q.FilterLess(field, t)

451 } else {

452 q.FilterGreater(field, t)

453 }

454 return nil

455 }

456 if f.Hold != nil {

457 q.FilterEqual("Hold", *f.Hold)

458 }

459 if f.Submitted != "" {

460 if err := applyTime("Queued", f.Submitted); err != nil {

461 return fmt.Errorf("applying filter for submitted: %v", err)

462 }

463 }

464 if f.NextAttempt != "" {

465 if err := applyTime("NextAttempt", f.NextAttempt); err != nil {

466 return fmt.Errorf("applying filter for next attempt: %v", err)

467 }

468 }

469 if f.Account != "" {

470 q.FilterNonzero(Msg{SenderAccount: f.Account})

471 }

472 if f.Transport != nil {

473 q.FilterEqual("Transport", *f.Transport)

474 }

475 if f.From != "" || f.To != "" {

476 q.FilterFn(func(m Msg) bool {

477 return f.From != "" && strings.Contains(m.Sender().XString(true), f.From) || f.To != "" && strings.Contains(m.Recipient().XString(true), f.To)

478 })

479 }

480 if f.Max != 0 {

481 q.Limit(f.Max)

482 }

483 return nil

484}

485

486type Sort struct {

487 Field string // "Queued" or "NextAttempt"/"".

488 LastID int64 // If > 0, we return objects beyond this, less/greater depending on Asc.

489 Last any // Value of Field for last object. Must be set iff LastID is set.

490 Asc bool // Ascending, or descending.

491}

492

493func (s Sort) apply(q *bstore.Query[Msg]) error {

494 switch s.Field {

495 case "", "NextAttempt":

496 s.Field = "NextAttempt"

497 case "Queued":

498 s.Field = "Queued"

499 default:

500 return fmt.Errorf("unknown sort order field %q", s.Field)

501 }

502

503 if s.LastID > 0 {

504 ls, ok := s.Last.(string)

505 if !ok {

506 return fmt.Errorf("last should be string with time, not %T %q", s.Last, s.Last)

507 }

508 last, err := time.Parse(time.RFC3339Nano, ls)

509 if err != nil {

510 last, err = time.Parse(time.RFC3339, ls)

511 }

512 if err != nil {

513 return fmt.Errorf("parsing last %q as time: %v", s.Last, err)

514 }

515 q.FilterNotEqual("ID", s.LastID)

516 var fieldEqual func(m Msg) bool

517 if s.Field == "NextAttempt" {

518 fieldEqual = func(m Msg) bool { return m.NextAttempt.Equal(last) }

519 } else {

520 fieldEqual = func(m Msg) bool { return m.Queued.Equal(last) }

521 }

522 if s.Asc {

523 q.FilterGreaterEqual(s.Field, last)

524 q.FilterFn(func(m Msg) bool {

525 return !fieldEqual(m) || m.ID > s.LastID

526 })

527 } else {

528 q.FilterLessEqual(s.Field, last)

529 q.FilterFn(func(m Msg) bool {

530 return !fieldEqual(m) || m.ID < s.LastID

531 })

532 }

533 }

534 if s.Asc {

535 q.SortAsc(s.Field, "ID")

536 } else {

537 q.SortDesc(s.Field, "ID")

538 }

539 return nil

540}

541

542// List returns max 100 messages matching filter in the delivery queue.

543// By default, orders by next delivery attempt.

544func List(ctx context.Context, filter Filter, sort Sort) ([]Msg, error) {

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

546 if err := filter.apply(q); err != nil {

547 return nil, err

548 }

549 if err := sort.apply(q); err != nil {

550 return nil, err

551 }

552 qmsgs, err := q.List()

553 if err != nil {

554 return nil, err

555 }

556 return qmsgs, nil

557}

558

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

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

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

562}

563

564// HoldRuleList returns all hold rules.

565func HoldRuleList(ctx context.Context) ([]HoldRule, error) {

566 return bstore.QueryDB[HoldRule](ctx, DB).List()

567}

568

569// HoldRuleAdd adds a new hold rule causing newly submitted messages to be marked

570// as "on hold", and existing matching messages too.

571func HoldRuleAdd(ctx context.Context, log mlog.Log, hr HoldRule) (HoldRule, error) {

572 var n int

573 err := DB.Write(ctx, func(tx *bstore.Tx) error {

574 hr.ID = 0

575 hr.SenderDomainStr = hr.SenderDomain.Name()

576 hr.RecipientDomainStr = hr.RecipientDomain.Name()

577 if err := tx.Insert(&hr); err != nil {

578 return err

579 }

580 log.Info("adding hold rule", slog.Any("holdrule", hr))

581

582 q := bstore.QueryTx[Msg](tx)

583 if !hr.All() {

584 q.FilterNonzero(Msg{

585 SenderAccount: hr.Account,

586 SenderDomainStr: hr.SenderDomainStr,

587 RecipientDomainStr: hr.RecipientDomainStr,

588 })

589 }

590 var err error

591 n, err = q.UpdateField("Hold", true)

592 if err != nil {

593 return fmt.Errorf("marking existing matching messages in queue on hold: %v", err)

594 }

595 return metricHoldUpdate(tx)

596 })

597 if err != nil {

598 return HoldRule{}, err

599 }

600 log.Info("marked messages in queue as on hold", slog.Int("messages", n))

601 msgqueueKick()

602 return hr, nil

603}

604

605// HoldRuleRemove removes a hold rule. The Hold field of existing messages are not

606// changed.

607func HoldRuleRemove(ctx context.Context, log mlog.Log, holdRuleID int64) error {

608 return DB.Write(ctx, func(tx *bstore.Tx) error {

609 hr := HoldRule{ID: holdRuleID}

610 if err := tx.Get(&hr); err != nil {

611 return err

612 }

613 log.Info("removing hold rule", slog.Any("holdrule", hr))

614 return tx.Delete(HoldRule{ID: holdRuleID})

615 })

616}

617

618// MakeMsg is a convenience function that sets the commonly used fields for a Msg.

619// messageID should include <>.

620func MakeMsg(sender, recipient smtp.Path, has8bit, smtputf8 bool, size int64, messageID string, prefix []byte, requireTLS *bool, next time.Time, subject string) Msg {

621 return Msg{

622 SenderLocalpart: sender.Localpart,

623 SenderDomain: sender.IPDomain,

624 RecipientLocalpart: recipient.Localpart,

625 RecipientDomain: recipient.IPDomain,

626 Has8bit: has8bit,

627 SMTPUTF8: smtputf8,

628 Size: size,

629 MessageID: messageID,

630 MsgPrefix: prefix,

631 Subject: subject,

632 RequireTLS: requireTLS,

633 Queued: time.Now(),

634 NextAttempt: next,

635 }

636}

637

638// Add one or more new messages to the queue. If the sender paths and MsgPrefix are

639// identical, they'll get the same BaseID, so they can be delivered in a single

640// SMTP transaction, with a single DATA command, but may be split into multiple

641// transactions if errors/limits are encountered. The queue is kicked immediately

642// to start a first delivery attempt.

643//

644// ID of the messagse must be 0 and will be set after inserting in the queue.

645//

646// Add sets derived fields like SenderDomainStr and RecipientDomainStr, and fields

647// related to queueing, such as Queued, NextAttempt.

648func Add(ctx context.Context, log mlog.Log, senderAccount string, msgFile *os.File, qml ...Msg) error {

649 if len(qml) == 0 {

650 return fmt.Errorf("must queue at least one message")

651 }

652

653 base := true

654

655 for i, qm := range qml {

656 if qm.ID != 0 {

657 return fmt.Errorf("id of queued messages must be 0")

658 }

659 // Sanity check, internal consistency.

660 qml[i].SenderDomainStr = formatIPDomain(qm.SenderDomain)

661 qml[i].RecipientDomainStr = formatIPDomain(qm.RecipientDomain)

662 if base && i > 0 && qm.Sender().String() != qml[0].Sender().String() || !bytes.Equal(qm.MsgPrefix, qml[0].MsgPrefix) {

663 base = false

664 }

665 }

666

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

668 if err != nil {

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

670 }

671 defer func() {

672 if tx != nil {

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

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

675 }

676 }

677 }()

678

679 // Mark messages Hold if they match a hold rule.

680 holdRules, err := bstore.QueryTx[HoldRule](tx).List()

681 if err != nil {

682 return fmt.Errorf("getting queue hold rules")

683 }

684

685 // Insert messages into queue. If multiple messages are to be delivered in a single

686 // transaction, they all get a non-zero BaseID that is the Msg.ID of the first

687 // message inserted.

688 var baseID int64

689 for i := range qml {

690 // FromIDs must be unique if present. We don't have a unique index because values

691 // can be the empty string. We check in both Msg and MsgRetired, both are relevant

692 // for uniquely identifying a message sent in the past.

693 if fromID := qml[i].FromID; fromID != "" {

694 if exists, err := bstore.QueryTx[Msg](tx).FilterNonzero(Msg{FromID: fromID}).Exists(); err != nil {

695 return fmt.Errorf("looking up fromid: %v", err)

696 } else if exists {

697 return fmt.Errorf("%w: fromid %q already present in message queue", ErrFromID, fromID)

698 }

699 if exists, err := bstore.QueryTx[MsgRetired](tx).FilterNonzero(MsgRetired{FromID: fromID}).Exists(); err != nil {

700 return fmt.Errorf("looking up fromid: %v", err)

701 } else if exists {

702 return fmt.Errorf("%w: fromid %q already present in retired message queue", ErrFromID, fromID)

703 }

704 }

705

706 qml[i].SenderAccount = senderAccount

707 qml[i].BaseID = baseID

708 for _, hr := range holdRules {

709 if hr.matches(qml[i]) {

710 qml[i].Hold = true

711 break

712 }

713 }

714 if err := tx.Insert(&qml[i]); err != nil {

715 return err

716 }

717 if base && i == 0 && len(qml) > 1 {

718 baseID = qml[i].ID

719 qml[i].BaseID = baseID

720 if err := tx.Update(&qml[i]); err != nil {

721 return err

722 }

723 }

724 }

725

726 var paths []string

727 defer func() {

728 for _, p := range paths {

729 err := os.Remove(p)

730 log.Check(err, "removing destination message file for queue", slog.String("path", p))

731 }

732 }()

733

734 syncDirs := map[string]struct{}{}

735

736 for _, qm := range qml {

737 dst := qm.MessagePath()

738 paths = append(paths, dst)

739

740 dstDir := filepath.Dir(dst)

741 if _, ok := syncDirs[dstDir]; !ok {

742 os.MkdirAll(dstDir, 0770)

743 syncDirs[dstDir] = struct{}{}

744 }

745

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

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

748 }

749 }

750

751 for dir := range syncDirs {

752 if err := moxio.SyncDir(log, dir); err != nil {

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

754 }

755 }

756

757 for _, m := range qml {

758 if m.Hold {

759 if err := metricHoldUpdate(tx); err != nil {

760 return err

761 }

762 break

763 }

764 }

765

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

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

768 }

769 tx = nil

770 paths = nil

771

772 msgqueueKick()

773

774 return nil

775}

776

777func formatIPDomain(d dns.IPDomain) string {

778 if len(d.IP) > 0 {

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

780 }

781 return d.Domain.Name()

782}

783

784var (

785 msgqueue = make(chan struct{}, 1)

786 deliveryResults = make(chan string, 1)

787)

788

789func kick() {

790 msgqueueKick()

791 hookqueueKick()

792}

793

794func msgqueueKick() {

795 select {

796 case msgqueue <- struct{}{}:

797 default:

798 }

799}

800

801// NextAttemptAdd adds a duration to the NextAttempt for all matching messages, and

802// kicks the queue.

803func NextAttemptAdd(ctx context.Context, filter Filter, d time.Duration) (affected int, rerr error) {

804 err := DB.Write(ctx, func(tx *bstore.Tx) error {

805 q := bstore.QueryTx[Msg](tx)

806 if err := filter.apply(q); err != nil {

807 return err

808 }

809 msgs, err := q.List()

810 if err != nil {

811 return fmt.Errorf("listing matching messages: %v", err)

812 }

813 for _, m := range msgs {

814 m.NextAttempt = m.NextAttempt.Add(d)

815 if err := tx.Update(&m); err != nil {

816 return err

817 }

818 }

819 affected = len(msgs)

820 return nil

821 })

822 if err != nil {

823 return 0, err

824 }

825 msgqueueKick()

826 return affected, nil

827}

828

829// NextAttemptSet sets NextAttempt for all matching messages to a new time, and

830// kicks the queue.

831func NextAttemptSet(ctx context.Context, filter Filter, t time.Time) (affected int, rerr error) {

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

833 if err := filter.apply(q); err != nil {

834 return 0, err

835 }

836 n, err := q.UpdateNonzero(Msg{NextAttempt: t})

837 if err != nil {

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

839 }

840 msgqueueKick()

841 return n, nil

842}

843

844// HoldSet sets Hold for all matching messages and kicks the queue.

845func HoldSet(ctx context.Context, filter Filter, hold bool) (affected int, rerr error) {

846 err := DB.Write(ctx, func(tx *bstore.Tx) error {

847 q := bstore.QueryTx[Msg](tx)

848 if err := filter.apply(q); err != nil {

849 return err

850 }

851 n, err := q.UpdateFields(map[string]any{"Hold": hold})

852 if err != nil {

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

854 }

855 affected = n

856 return metricHoldUpdate(tx)

857 })

858 if err != nil {

859 return 0, err

860 }

861 msgqueueKick()

862 return affected, nil

863}

864

865// TransportSet changes the transport to use for the matching messages.

866func TransportSet(ctx context.Context, filter Filter, transport string) (affected int, rerr error) {

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

868 if err := filter.apply(q); err != nil {

869 return 0, err

870 }

871 n, err := q.UpdateFields(map[string]any{"Transport": transport})

872 if err != nil {

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

874 }

875 msgqueueKick()

876 return n, nil

877}

878

879// Fail marks matching messages as failed for delivery, delivers a DSN to the

880// sender, and sends a webhook.

881//

882// Returns number of messages removed, which can be non-zero even in case of an

883// error.

884func Fail(ctx context.Context, log mlog.Log, f Filter) (affected int, err error) {

885 return failDrop(ctx, log, f, true)

886}

887

888// Drop removes matching messages from the queue. Messages are added as retired

889// message, webhooks with the "canceled" event are queued.

890//

891// Returns number of messages removed, which can be non-zero even in case of an

892// error.

893func Drop(ctx context.Context, log mlog.Log, f Filter) (affected int, err error) {

894 return failDrop(ctx, log, f, false)

895}

896

897func failDrop(ctx context.Context, log mlog.Log, filter Filter, fail bool) (affected int, rerr error) {

898 var msgs []Msg

899 err := DB.Write(ctx, func(tx *bstore.Tx) error {

900 q := bstore.QueryTx[Msg](tx)

901 if err := filter.apply(q); err != nil {

902 return err

903 }

904 var err error

905 msgs, err = q.List()

906 if err != nil {

907 return fmt.Errorf("getting messages to delete: %v", err)

908 }

909

910 if len(msgs) == 0 {

911 return nil

912 }

913

914 now := time.Now()

915 var remoteMTA dsn.NameIP

916 for i := range msgs {

917 result := MsgResult{

918 Start: now,

919 Error: "delivery canceled by admin",

920 }

921 msgs[i].Results = append(msgs[i].Results, result)

922 if fail {

923 if msgs[i].LastAttempt == nil {

924 msgs[i].LastAttempt = &now

925 }

926 deliverDSNFailure(log, msgs[i], remoteMTA, "", result.Error, nil)

927 }

928 }

929 event := webhook.EventCanceled

930 if fail {

931 event = webhook.EventFailed

932 }

933 if err := retireMsgs(log, tx, event, 0, "", nil, msgs...); err != nil {

934 return fmt.Errorf("removing queue messages from database: %w", err)

935 }

936 return metricHoldUpdate(tx)

937 })

938 if err != nil {

939 return 0, err

940 }

941 if len(msgs) > 0 {

942 if err := removeMsgsFS(log, msgs...); err != nil {

943 return len(msgs), fmt.Errorf("removing queue messages from file system: %w", err)

944 }

945 }

946 kick()

947 return len(msgs), nil

948}

949

950// RequireTLSSet updates the RequireTLS field of matching messages.

951func RequireTLSSet(ctx context.Context, filter Filter, requireTLS *bool) (affected int, rerr error) {

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

953 if err := filter.apply(q); err != nil {

954 return 0, err

955 }

956 n, err := q.UpdateFields(map[string]any{"RequireTLS": requireTLS})

957 msgqueueKick()

958 return n, err

959}

960

961// RetiredFilter filters messages to list or operate on. Used by admin web interface

962// and cli.

963//

964// Only non-empty/non-zero values are applied to the filter. Leaving all fields

965// empty/zero matches all messages.

966type RetiredFilter struct {

967 Max int

968 IDs []int64

969 Account string

970 From string

971 To string

972 Submitted string // Whether submitted before/after a time relative to now. ">$duration" or "<$duration", also with "now" for duration.

973 LastActivity string // ">$duration" or "<$duration", also with "now" for duration.

974 Transport *string

975 Success *bool

976}

977

978func (f RetiredFilter) apply(q *bstore.Query[MsgRetired]) error {

979 if len(f.IDs) > 0 {

980 q.FilterIDs(f.IDs)

981 }

982 applyTime := func(field string, s string) error {

983 orig := s

984 var before bool

985 if strings.HasPrefix(s, "<") {

986 before = true

987 } else if !strings.HasPrefix(s, ">") {

988 return fmt.Errorf(`must start with "<" for before or ">" for after a duration`)

989 }

990 s = strings.TrimSpace(s[1:])

991 var t time.Time

992 if s == "now" {

993 t = time.Now()

994 } else if d, err := time.ParseDuration(s); err != nil {

995 return fmt.Errorf("parsing duration %q: %v", orig, err)

996 } else {

997 t = time.Now().Add(d)

998 }

999 if before {

1000 q.FilterLess(field, t)

1001 } else {

1002 q.FilterGreater(field, t)

1003 }

1004 return nil

1005 }

1006 if f.Submitted != "" {

1007 if err := applyTime("Queued", f.Submitted); err != nil {

1008 return fmt.Errorf("applying filter for submitted: %v", err)

1009 }

1010 }

1011 if f.LastActivity != "" {

1012 if err := applyTime("LastActivity", f.LastActivity); err != nil {

1013 return fmt.Errorf("applying filter for last activity: %v", err)

1014 }

1015 }

1016 if f.Account != "" {

1017 q.FilterNonzero(MsgRetired{SenderAccount: f.Account})

1018 }

1019 if f.Transport != nil {

1020 q.FilterEqual("Transport", *f.Transport)

1021 }

1022 if f.From != "" || f.To != "" {

1023 q.FilterFn(func(m MsgRetired) bool {

1024 return f.From != "" && strings.Contains(m.SenderLocalpart.String()+"@"+m.SenderDomainStr, f.From) || f.To != "" && strings.Contains(m.Recipient().XString(true), f.To)

1025 })

1026 }

1027 if f.Success != nil {

1028 q.FilterEqual("Success", *f.Success)

1029 }

1030 if f.Max != 0 {

1031 q.Limit(f.Max)

1032 }

1033 return nil

1034}

1035

1036type RetiredSort struct {

1037 Field string // "Queued" or "LastActivity"/"".

1038 LastID int64 // If > 0, we return objects beyond this, less/greater depending on Asc.

1039 Last any // Value of Field for last object. Must be set iff LastID is set.

1040 Asc bool // Ascending, or descending.

1041}

1042

1043func (s RetiredSort) apply(q *bstore.Query[MsgRetired]) error {

1044 switch s.Field {

1045 case "", "LastActivity":

1046 s.Field = "LastActivity"

1047 case "Queued":

1048 s.Field = "Queued"

1049 default:

1050 return fmt.Errorf("unknown sort order field %q", s.Field)

1051 }

1052

1053 if s.LastID > 0 {

1054 ls, ok := s.Last.(string)

1055 if !ok {

1056 return fmt.Errorf("last should be string with time, not %T %q", s.Last, s.Last)

1057 }

1058 last, err := time.Parse(time.RFC3339Nano, ls)

1059 if err != nil {

1060 last, err = time.Parse(time.RFC3339, ls)

1061 }

1062 if err != nil {

1063 return fmt.Errorf("parsing last %q as time: %v", s.Last, err)

1064 }

1065 q.FilterNotEqual("ID", s.LastID)

1066 var fieldEqual func(m MsgRetired) bool

1067 if s.Field == "LastActivity" {

1068 fieldEqual = func(m MsgRetired) bool { return m.LastActivity.Equal(last) }

1069 } else {

1070 fieldEqual = func(m MsgRetired) bool { return m.Queued.Equal(last) }

1071 }

1072 if s.Asc {

1073 q.FilterGreaterEqual(s.Field, last)

1074 q.FilterFn(func(mr MsgRetired) bool {

1075 return !fieldEqual(mr) || mr.ID > s.LastID

1076 })

1077 } else {

1078 q.FilterLessEqual(s.Field, last)

1079 q.FilterFn(func(mr MsgRetired) bool {

1080 return !fieldEqual(mr) || mr.ID < s.LastID

1081 })

1082 }

1083 }

1084 if s.Asc {

1085 q.SortAsc(s.Field, "ID")

1086 } else {

1087 q.SortDesc(s.Field, "ID")

1088 }

1089 return nil

1090}

1091

1092// RetiredList returns retired messages.

1093func RetiredList(ctx context.Context, filter RetiredFilter, sort RetiredSort) ([]MsgRetired, error) {

1094 q := bstore.QueryDB[MsgRetired](ctx, DB)

1095 if err := filter.apply(q); err != nil {

1096 return nil, err

1097 }

1098 if err := sort.apply(q); err != nil {

1099 return nil, err

1100 }

1101 return q.List()

1102}

1103

1104type ReadReaderAtCloser interface {

1105 io.ReadCloser

1106 io.ReaderAt

1107}

1108

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

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

1111 qm := Msg{ID: id}

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

1113 if err != nil {

1114 return nil, err

1115 }

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

1117 if err != nil {

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

1119 }

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

1121 return r, err

1122}

1123

1124const maxConcurrentDeliveries = 10

1125const maxConcurrentHookDeliveries = 10

1126

1127// Start opens the database by calling Init, then starts the delivery and cleanup

1128// processes.

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

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

1131 return err

1132 }

1133

1134 go startQueue(resolver, done)

1135 go startHookQueue(done)

1136

1137 go cleanupMsgRetired(done)

1138 go cleanupHookRetired(done)

1139

1140 return nil

1141}

1142

1143func cleanupMsgRetired(done chan struct{}) {

1144 log := mlog.New("queue", nil)

1145

1146 defer func() {

1147 x := recover()

1148 if x != nil {

1149 log.Error("unhandled panic in cleanupMsgRetired", slog.Any("x", x))

1150 debug.PrintStack()

1151 metrics.PanicInc(metrics.Queue)

1152 }

1153 }()

1154

1155 timer := time.NewTimer(3 * time.Second)

1156 for {

1157 select {

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

1159 done <- struct{}{}

1160 return

1161 case <-timer.C:

1162 }

1163

1164 cleanupMsgRetiredSingle(log)

1165 timer.Reset(time.Hour)

1166 }

1167}

1168

1169func cleanupMsgRetiredSingle(log mlog.Log) {

1170 n, err := bstore.QueryDB[MsgRetired](mox.Shutdown, DB).FilterLess("KeepUntil", time.Now()).Delete()

1171 log.Check(err, "removing old retired messages")

1172 if n > 0 {

1173 log.Debug("cleaned up retired messages", slog.Int("count", n))

1174 }

1175}

1176

1177func startQueue(resolver dns.Resolver, done chan struct{}) {

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

1179 log := mlog.New("queue", nil)

1180

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

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

1183

1184 timer := time.NewTimer(0)

1185

1186 for {

1187 select {

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

1189 for len(busyDomains) > 0 {

1190 domain := <-deliveryResults

1191 delete(busyDomains, domain)

1192 }

1193 done <- struct{}{}

1194 return

1195 case <-msgqueue:

1196 case <-timer.C:

1197 case domain := <-deliveryResults:

1198 delete(busyDomains, domain)

1199 }

1200

1201 if len(busyDomains) >= maxConcurrentDeliveries {

1202 continue

1203 }

1204

1205 launchWork(log, resolver, busyDomains)

1206 timer.Reset(nextWork(mox.Shutdown, log, busyDomains))

1207 }

1208}

1209

1210func nextWork(ctx context.Context, log mlog.Log, busyDomains map[string]struct{}) time.Duration {

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

1212 if len(busyDomains) > 0 {

1213 var doms []any

1214 for d := range busyDomains {

1215 doms = append(doms, d)

1216 }

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

1218 }

1219 q.FilterEqual("Hold", false)

1220 q.SortAsc("NextAttempt")

1221 q.Limit(1)

1222 qm, err := q.Get()

1223 if err == bstore.ErrAbsent {

1224 return 24 * time.Hour

1225 } else if err != nil {

1226 log.Errorx("finding time for next delivery attempt", err)

1227 return 1 * time.Minute

1228 }

1229 return time.Until(qm.NextAttempt)

1230}

1231

1232func launchWork(log mlog.Log, resolver dns.Resolver, busyDomains map[string]struct{}) int {

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

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

1235 q.FilterEqual("Hold", false)

1236 q.SortAsc("NextAttempt")

1237 q.Limit(maxConcurrentDeliveries)

1238 if len(busyDomains) > 0 {

1239 var doms []any

1240 for d := range busyDomains {

1241 doms = append(doms, d)

1242 }

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

1244 }

1245 var msgs []Msg

1246 seen := map[string]bool{}

1247 err := q.ForEach(func(m Msg) error {

1248 dom := m.RecipientDomainStr

1249 if _, ok := busyDomains[dom]; !ok && !seen[dom] {

1250 seen[dom] = true

1251 msgs = append(msgs, m)

1252 }

1253 return nil

1254 })

1255 if err != nil {

1256 log.Errorx("querying for work in queue", err)

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

1258 return -1

1259 }

1260

1261 for _, m := range msgs {

1262 busyDomains[m.RecipientDomainStr] = struct{}{}

1263 go deliver(log, resolver, m)

1264 }

1265 return len(msgs)

1266}

1267

1268// 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.

1269

1270func removeMsgsFS(log mlog.Log, msgs ...Msg) error {

1271 var errs []string

1272 for _, m := range msgs {

1273 p := mox.DataDirPath(filepath.Join("queue", store.MessagePath(m.ID)))

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

1275 errs = append(errs, fmt.Sprintf("%s: %v", p, err))

1276 }

1277 }

1278 if len(errs) > 0 {

1279 return fmt.Errorf("removing message files from queue: %s", strings.Join(errs, "; "))

1280 }

1281 return nil

1282}

1283

1284// Move one or more messages to retire list or remove it. Webhooks are scheduled.

1285// IDs of msgs in suppressedMsgIDs caused a suppression to be added.

1286//

1287// Callers should update Msg.Results before calling.

1288//

1289// Callers must remove the messages from the file system afterwards, see

1290// removeMsgsFS. Callers must also kick the message and webhook queues.

1291func retireMsgs(log mlog.Log, tx *bstore.Tx, event webhook.OutgoingEvent, code int, secode string, suppressedMsgIDs []int64, msgs ...Msg) error {

1292 now := time.Now()

1293

1294 var hooks []Hook

1295 m0 := msgs[0]

1296 accConf, ok := mox.Conf.Account(m0.SenderAccount)

1297 var hookURL string

1298 if accConf.OutgoingWebhook != nil {

1299 hookURL = accConf.OutgoingWebhook.URL

1300 }

1301 log.Debug("retiring messages from queue", slog.Any("event", event), slog.String("account", m0.SenderAccount), slog.Bool("ok", ok), slog.String("webhookurl", hookURL))

1302 if hookURL != "" && (len(accConf.OutgoingWebhook.Events) == 0 || slices.Contains(accConf.OutgoingWebhook.Events, string(event))) {

1303 for _, m := range msgs {

1304 suppressing := slices.Contains(suppressedMsgIDs, m.ID)

1305 h, err := hookCompose(m, hookURL, accConf.OutgoingWebhook.Authorization, event, suppressing, code, secode)

1306 if err != nil {

1307 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()))

1308 } else {

1309 hooks = append(hooks, h)

1310 }

1311 }

1312 }

1313

1314 msgKeep := 24 * 7 * time.Hour

1315 hookKeep := 24 * 7 * time.Hour

1316 if ok {

1317 msgKeep = accConf.KeepRetiredMessagePeriod

1318 hookKeep = accConf.KeepRetiredWebhookPeriod

1319 }

1320

1321 for _, m := range msgs {

1322 if err := tx.Delete(&m); err != nil {

1323 return err

1324 }

1325 }

1326 if msgKeep > 0 {

1327 for _, m := range msgs {

1328 rm := m.Retired(event == webhook.EventDelivered, now, now.Add(msgKeep))

1329 if err := tx.Insert(&rm); err != nil {

1330 return err

1331 }

1332 }

1333 }

1334

1335 for i := range hooks {

1336 if err := hookInsert(tx, &hooks[i], now, hookKeep); err != nil {

1337 return fmt.Errorf("enqueueing webhooks while retiring messages from queue: %v", err)

1338 }

1339 }

1340

1341 if len(hooks) > 0 {

1342 for _, h := range hooks {

1343 log.Debug("queued webhook while retiring message from queue", h.attrs()...)

1344 }

1345 hookqueueKick()

1346 }

1347 return nil

1348}

1349

1350// deliver attempts to deliver a message.

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

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

1353func deliver(log mlog.Log, resolver dns.Resolver, m0 Msg) {

1354 ctx := mox.Shutdown

1355

1356 qlog := log.WithCid(mox.Cid()).With(

1357 slog.Any("from", m0.Sender()),

1358 slog.Int("attempts", m0.Attempts+1))

1359

1360 defer func() {

1361 deliveryResults <- formatIPDomain(m0.RecipientDomain)

1362

1363 x := recover()

1364 if x != nil {

1365 qlog.Error("deliver panic", slog.Any("panic", x), slog.Int64("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))

1366 debug.PrintStack()

1367 metrics.PanicInc(metrics.Queue)

1368 }

1369 }()

1370

1371 // We'll use a single transaction for the various checks, committing as soon as

1372 // we're done with it.

1373 xtx, err := DB.Begin(mox.Shutdown, true)

1374 if err != nil {

1375 qlog.Errorx("transaction for gathering messages to deliver", err)

1376 return

1377 }

1378 defer func() {

1379 if xtx != nil {

1380 err := xtx.Rollback()

1381 qlog.Check(err, "rolling back transaction after error delivering")

1382 }

1383 }()

1384

1385 // We register this attempt by setting LastAttempt, adding an empty Result, and

1386 // already setting NextAttempt in the future with exponential backoff. If we run

1387 // into trouble delivery below, at least we won't be bothering the receiving server

1388 // with our problems.

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

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

1391 // ../rfc/5321:3703

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

1393 now := time.Now()

1394 var backoff time.Duration

1395 var origNextAttempt time.Time

1396 prepare := func() error {

1397 // Refresh message within transaction.

1398 m0 = Msg{ID: m0.ID}

1399 if err := xtx.Get(&m0); err != nil {

1400 return fmt.Errorf("get message to be delivered: %v", err)

1401 }

1402

1403 backoff = time.Duration(7*60+30+jitter.IntN(10)-5) * time.Second

1404 for range m0.Attempts {

1405 backoff *= time.Duration(2)

1406 }

1407 m0.Attempts++

1408 origNextAttempt = m0.NextAttempt

1409 m0.LastAttempt = &now

1410 m0.NextAttempt = now.Add(backoff)

1411 m0.Results = append(m0.Results, MsgResult{Start: now, Error: resultErrorDelivering})

1412 if err := xtx.Update(&m0); err != nil {

1413 return fmt.Errorf("update message to be delivered: %v", err)

1414 }

1415 return nil

1416 }

1417 if err := prepare(); err != nil {

1418 qlog.Errorx("storing delivery attempt", err, slog.Int64("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))

1419 return

1420 }

1421

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

1423

1424 // If domain of sender is currently disabled, fail the delivery attempt.

1425 if domConf, _ := mox.Conf.Domain(m0.SenderDomain.Domain); domConf.Disabled {

1426 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, fmt.Errorf("domain of sender temporarily disabled"))

1427 err = xtx.Commit()

1428 qlog.Check(err, "commit processing failure to deliver messages")

1429 xtx = nil

1430 kick()

1431 return

1432 }

1433

1434 // Check if recipient is on suppression list. If so, fail delivery.

1435 path := smtp.Path{Localpart: m0.RecipientLocalpart, IPDomain: m0.RecipientDomain}

1436 baseAddr := baseAddress(path).XString(true)

1437 qsup := bstore.QueryTx[webapi.Suppression](xtx)

1438 qsup.FilterNonzero(webapi.Suppression{Account: m0.SenderAccount, BaseAddress: baseAddr})

1439 exists, err := qsup.Exists()

1440 if err != nil || exists {

1441 if err != nil {

1442 qlog.Errorx("checking whether recipient address is in suppression list", err)

1443 } else {

1444 err := fmt.Errorf("not delivering to recipient address %s: %w", path.XString(true), errSuppressed)

1445 err = smtpclient.Error{Permanent: true, Err: err}

1446 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, err)

1447 }

1448 err = xtx.Commit()

1449 qlog.Check(err, "commit processing failure to deliver messages")

1450 xtx = nil

1451 kick()

1452 return

1453 }

1454

1455 resolveTransport := func(mm Msg) (string, config.Transport, bool) {

1456 if mm.Transport != "" {

1457 transport, ok := mox.Conf.Static.Transports[mm.Transport]

1458 if !ok {

1459 return "", config.Transport{}, false

1460 }

1461 return mm.Transport, transport, ok

1462 }

1463 route := findRoute(mm.Attempts, mm)

1464 return route.Transport, route.ResolvedTransport, true

1465 }

1466

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

1468 m0.Attempts--

1469 transportName, transport, transportOK := resolveTransport(m0)

1470 m0.Attempts++

1471 if !transportOK {

1472 failMsgsTx(qlog, xtx, []*Msg{&m0}, m0.DialedIPs, backoff, remoteMTA, fmt.Errorf("cannot find transport %q", m0.Transport))

1473 err = xtx.Commit()

1474 qlog.Check(err, "commit processing failure to deliver messages")

1475 xtx = nil

1476 kick()

1477 return

1478 }

1479

1480 if transportName != "" {

1481 qlog = qlog.With(slog.String("transport", transportName))

1482 qlog.Debug("delivering with transport")

1483 }

1484

1485 // Attempt to gather more recipients for this identical message, only with the same

1486 // recipient domain, and under the same conditions (recipientdomain, attempts,

1487 // requiretls, transport). ../rfc/5321:3759

1488 msgs := []*Msg{&m0}

1489 if m0.BaseID != 0 {

1490 gather := func() error {

1491 q := bstore.QueryTx[Msg](xtx)

1492 q.FilterNonzero(Msg{BaseID: m0.BaseID, RecipientDomainStr: m0.RecipientDomainStr, Attempts: m0.Attempts - 1})

1493 q.FilterNotEqual("ID", m0.ID)

1494 q.FilterLessEqual("NextAttempt", origNextAttempt)

1495 q.FilterEqual("Hold", false)

1496 err := q.ForEach(func(xm Msg) error {

1497 mrtls := m0.RequireTLS != nil

1498 xmrtls := xm.RequireTLS != nil

1499 if mrtls != xmrtls || mrtls && *m0.RequireTLS != *xm.RequireTLS {

1500 return nil

1501 }

1502 tn, _, ok := resolveTransport(xm)

1503 if ok && tn == transportName {

1504 msgs = append(msgs, &xm)

1505 }

1506 return nil

1507 })

1508 if err != nil {

1509 return fmt.Errorf("looking up more recipients: %v", err)

1510 }

1511

1512 // Mark these additional messages as attempted too.

1513 for _, mm := range msgs[1:] {

1514 mm.Attempts++

1515 mm.NextAttempt = m0.NextAttempt

1516 mm.LastAttempt = m0.LastAttempt

1517 mm.Results = append(mm.Results, MsgResult{Start: now, Error: resultErrorDelivering})

1518 if err := xtx.Update(mm); err != nil {

1519 return fmt.Errorf("updating more message recipients for smtp transaction: %v", err)

1520 }

1521 }

1522 return nil

1523 }

1524 if err := gather(); err != nil {

1525 qlog.Errorx("error finding more recipients for message, will attempt to send to single recipient", err)

1526 msgs = msgs[:1]

1527 }

1528 }

1529

1530 if err := xtx.Commit(); err != nil {

1531 qlog.Errorx("commit of preparation to deliver", err, slog.Any("msgid", m0.ID))

1532 return

1533 }

1534 xtx = nil

1535

1536 if len(msgs) > 1 {

1537 ids := make([]int64, len(msgs))

1538 rcpts := make([]smtp.Path, len(msgs))

1539 for i, m := range msgs {

1540 ids[i] = m.ID

1541 rcpts[i] = m.Recipient()

1542 }

1543 qlog.Debug("delivering to multiple recipients", slog.Any("msgids", ids), slog.Any("recipients", rcpts))

1544 } else {

1545 qlog.Debug("delivering to single recipient", slog.Any("msgid", m0.ID), slog.Any("recipient", m0.Recipient()))

1546 }

1547

1548 // Test for "Fail" transport before Localserve.

1549 if transport.Fail != nil {

1550 err := smtpclient.Error{

1551 Permanent: transport.Fail.Code/100 == 5,

1552 Code: transport.Fail.Code,

1553 Secode: smtp.SePol7Other0,

1554 Err: fmt.Errorf("%s", transport.Fail.Message),

1555 }

1556 failMsgsDB(qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, err)

1557 return

1558 }

1559

1560 if Localserve {

1561 deliverLocalserve(ctx, qlog, msgs, backoff)

1562 return

1563 }

1564

1565 // We gather TLS connection successes and failures during delivery, and we store

1566 // them in tlsrptdb. Every 24 hours we send an email with a report to the recipient

1567 // domains that opt in via a TLSRPT DNS record. For us, the tricky part is

1568 // collecting all reporting information. We've got several TLS modes

1569 // (opportunistic, DANE and/or MTA-STS (PKIX), overrides due to Require TLS).

1570 // Failures can happen at various levels: MTA-STS policies (apply to whole delivery

1571 // attempt/domain), MX targets (possibly multiple per delivery attempt, both for

1572 // MTA-STS and DANE).

1573 //

1574 // Once the SMTP client has tried a TLS handshake, we register success/failure,

1575 // regardless of what happens next on the connection. We also register failures

1576 // when they happen before we get to the SMTP client, but only if they are related

1577 // to TLS (and some DNSSEC).

1578 var recipientDomainResult tlsrpt.Result

1579 var hostResults []tlsrpt.Result

1580 defer func() {

1581 if mox.Conf.Static.NoOutgoingTLSReports || m0.RecipientDomain.IsIP() {

1582 return

1583 }

1584

1585 now := time.Now()

1586 dayUTC := now.UTC().Format("20060102")

1587

1588 // See if this contains a failure. If not, we'll mark TLS results for delivering

1589 // DMARC reports SendReport false, so we won't as easily get into a report sending

1590 // loop.

1591 var failure bool

1592 for _, result := range hostResults {

1593 if result.Summary.TotalFailureSessionCount > 0 {

1594 failure = true

1595 break

1596 }

1597 }

1598 if recipientDomainResult.Summary.TotalFailureSessionCount > 0 {

1599 failure = true

1600 }

1601

1602 results := make([]tlsrptdb.TLSResult, 0, 1+len(hostResults))

1603 tlsaPolicyDomains := map[string]bool{}

1604 addResult := func(r tlsrpt.Result, isHost bool) {

1605 var zerotype tlsrpt.PolicyType

1606 if r.Policy.Type == zerotype {

1607 return

1608 }

1609

1610 // Ensure we store policy domain in unicode in database.

1611 policyDomain, err := dns.ParseDomain(r.Policy.Domain)

1612 if err != nil {

1613 qlog.Errorx("parsing policy domain for tls result", err, slog.String("policydomain", r.Policy.Domain))

1614 return

1615 }

1616

1617 if r.Policy.Type == tlsrpt.TLSA {

1618 tlsaPolicyDomains[policyDomain.ASCII] = true

1619 }

1620

1621 tlsResult := tlsrptdb.TLSResult{

1622 PolicyDomain: policyDomain.Name(),

1623 DayUTC: dayUTC,

1624 RecipientDomain: m0.RecipientDomain.Domain.Name(),

1625 IsHost: isHost,

1626 SendReport: !m0.IsTLSReport && (!m0.IsDMARCReport || failure),

1627 Results: []tlsrpt.Result{r},

1628 }

1629 results = append(results, tlsResult)

1630 }

1631 for _, result := range hostResults {

1632 addResult(result, true)

1633 }

1634 // If we were delivering to a mail host directly (not a domain with MX records), we

1635 // are more likely to get a TLSA policy than an STS policy. Don't potentially

1636 // confuse operators with both a tlsa and no-policy-found result.

1637 // todo spec: ../rfc/8460:440 an explicit no-sts-policy result would be useful.

1638 if recipientDomainResult.Policy.Type != tlsrpt.NoPolicyFound || !tlsaPolicyDomains[recipientDomainResult.Policy.Domain] {

1639 addResult(recipientDomainResult, false)

1640 }

1641

1642 if len(results) > 0 {

1643 err := tlsrptdb.AddTLSResults(context.Background(), results)

1644 qlog.Check(err, "adding tls results to database for upcoming tlsrpt report")

1645 }

1646 }()

1647

1648 var dialer smtpclient.Dialer = &net.Dialer{}

1649 if transport.Submissions != nil {

1650 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submissions, true, 465)

1651 } else if transport.Submission != nil {

1652 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.Submission, false, 587)

1653 } else if transport.SMTP != nil {

1654 // todo future: perhaps also gather tlsrpt results for submissions.

1655 deliverSubmit(qlog, resolver, dialer, msgs, backoff, transportName, transport.SMTP, false, 25)

1656 } else {

1657 ourHostname := mox.Conf.Static.HostnameDomain

1658 if transport.Socks != nil {

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

1660 if err != nil {

1661 failMsgsDB(qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer: %v", err))

1662 return

1663 } else if d, ok := socksdialer.(smtpclient.Dialer); !ok {

1664 failMsgsDB(qlog, msgs, msgs[0].DialedIPs, backoff, dsn.NameIP{}, fmt.Errorf("socks dialer is not a contextdialer"))

1665 return

1666 } else {

1667 dialer = d

1668 }

1669 ourHostname = transport.Socks.Hostname

1670 }

1671 recipientDomainResult, hostResults = deliverDirect(qlog, resolver, dialer, ourHostname, transportName, transport.Direct, msgs, backoff)

1672 }

1673}

1674

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

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

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

1678 return r

1679 }

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

1681 return r

1682 }

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

1684 return r

1685 }

1686 return config.Route{}

1687}

1688

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

1690 for _, r := range routes {

1691 if routeMatch(attempt, m, r) {

1692 return r, true

1693 }

1694 }

1695 return config.Route{}, false

1696}

1697

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

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

1700}

1701

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

1703 if len(l) == 0 {

1704 return true

1705 }

1706 for _, e := range l {

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

1708 return true

1709 }

1710 }

1711 return false