From 4be088ca3a62cdd4febde8dd91a20efd4d9e13e3 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 9 Oct 2026 12:55:57 +0000 Subject: [PATCH] feat: make IMAP hydration retryable, resume bounded completed-folder generation recovery, honor NOMODSEQ and bound cancellable TLS session setup Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- go.mod | 2 +- go.sum | 2 + internal/app/worker/wmail/imap_conn.go | 2 +- internal/app/worker/wmail/sync_imap.go | 87 +++- .../app/worker/wmail/sync_imap_recovery.go | 146 ++++++ .../worker/wmail/sync_imap_recovery_test.go | 229 ++++++++ internal/app/worker/wmail/sync_imap_test.go | 34 +- internal/client/smtpimap/imap/body_read.go | 44 +- internal/client/smtpimap/imap/client.go | 115 +++- .../client/smtpimap/imap/fetch_caps_test.go | 4 + internal/client/smtpimap/imap/search_date.go | 1 + .../smtpimap/imap/sync_repair_wire_test.go | 490 ++++++++++++++++++ 12 files changed, 1099 insertions(+), 57 deletions(-) create mode 100644 internal/app/worker/wmail/sync_imap_recovery.go create mode 100644 internal/app/worker/wmail/sync_imap_recovery_test.go create mode 100644 internal/client/smtpimap/imap/sync_repair_wire_test.go diff --git a/go.mod b/go.mod index 2b1aa2f15..551e34ae1 100644 --- a/go.mod +++ b/go.mod @@ -24,7 +24,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/ssm v1.67.7 github.com/aws/smithy-go v1.28.1 github.com/confluentinc/confluent-kafka-go/v2 v2.12.0 - github.com/emersion/go-imap/v2 v2.0.0-beta.5 + github.com/emersion/go-imap/v2 v2.0.0-beta.7 github.com/emersion/go-message v0.18.1 github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 github.com/getsentry/sentry-go v0.36.2 diff --git a/go.sum b/go.sum index daee0c86c..f4a37ffcc 100644 --- a/go.sum +++ b/go.sum @@ -269,6 +269,8 @@ github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203 h1:XBBHcIb256gUJ github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203/go.mod h1:E1jcSv8FaEny+OP/5k9UxZVw9YFWGj7eI4KR/iOBqCg= github.com/emersion/go-imap/v2 v2.0.0-beta.5 h1:H3858DNmBuXyMK1++YrQIRdpKE1MwBc+ywBtg3n+0wA= github.com/emersion/go-imap/v2 v2.0.0-beta.5/go.mod h1:BZTFHsS1hmgBkFlHqbxGLXk2hnRqTItUgwjSSCsYNAk= +github.com/emersion/go-imap/v2 v2.0.0-beta.7 h1:lNznYWa5uhMrngnSYEklzCeye4DBq9TEJ+pr0K593+8= +github.com/emersion/go-imap/v2 v2.0.0-beta.7/go.mod h1:BZTFHsS1hmgBkFlHqbxGLXk2hnRqTItUgwjSSCsYNAk= github.com/emersion/go-message v0.18.1 h1:tfTxIoXFSFRwWaZsgnqS1DSZuGpYGzSmCZD8SK3QA2E= github.com/emersion/go-message v0.18.1/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA= github.com/emersion/go-msgauth v0.7.0 h1:vj2hMn6KhFtW41kshIBTXvp6KgYSqpA/ZN9Pv4g1INc= diff --git a/internal/app/worker/wmail/imap_conn.go b/internal/app/worker/wmail/imap_conn.go index bbf0dee7f..3e36c4a24 100644 --- a/internal/app/worker/wmail/imap_conn.go +++ b/internal/app/worker/wmail/imap_conn.go @@ -40,7 +40,7 @@ type ImapConn interface { FetchFlags(ctx context.Context, uidFrom uint32) (map[uint32]imap.FlagState, *errx.MailError) SearchSince(since time.Time) ([]goimap.UID, *errx.MailError) FetchEnvelopes(ctx context.Context, uids []goimap.UID) ([]*imap.Fetched, *errx.MailError) - FetchBody(f *imap.Fetched) + FetchBody(f *imap.Fetched) *errx.MailError // Send path. AppendToSent(ctx context.Context, raw []byte, sentAt time.Time) error diff --git a/internal/app/worker/wmail/sync_imap.go b/internal/app/worker/wmail/sync_imap.go index bed4867f2..a84ff11ee 100644 --- a/internal/app/worker/wmail/sync_imap.go +++ b/internal/app/worker/wmail/sync_imap.go @@ -76,14 +76,12 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { return nil } - // condStore decides the incremental strategy for the whole account: - // mod-sequences where the server has CONDSTORE, UIDNEXT where it does not - // (Outlook.com, Microsoft 365 over IMAP, Yahoo, many hosted servers). - condStore := client.HasCondStore() + // CONDSTORE is effective per folder; NOMODSEQ uses UIDNEXT and periodic flags. caughtUp := true for i := range folders { box := &folders[i] + condStore := client.HasCondStore() && box.HighestModSeq > 0 // The listing is remembered before anything is decided from it, so // the departure check below reads the previous pass, never this one. prevListing, listedBefore := w.listed[box.Name] @@ -93,6 +91,10 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { // First sight: baseline. Live sync starts from this cursor; the // backfill owns everything before it. saved := *box + if !w.imapRecoverFolder(box) { + stats.aborted = true + return nil + } if err := w.mboxEvent(&saved); err != nil { return nil } @@ -105,20 +107,38 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { // snapshot is about messages that may no longer be there. Re-baseline // it exactly like a first sighting, and drop its backfill floor so an // import still running walks it again (stored messages are matched by - // Message-ID, so nothing is stored twice). A finished import stays - // finished; the mail that is already here keeps its rows, and its - // stale UIDs are what the warmup path checks the generation against. + // Message-ID, so nothing is stored twice). Completed accounts schedule + // a separate bounded folder recovery without resetting global history. if befBox.UIDValidity != box.UIDValidity { saved := *box + if w.tracker.state.BackfillStatus != models.SyncBackfillComplete { + w.tracker.setFolder(box.Name, models.SyncFolderCursor{}) + } + if !w.imapRecoverFolder(box) { + stats.aborted = true + return nil + } if err := w.mboxEvent(&saved); err != nil { return nil } *befBox = saved delete(w.flagScan, box.Name) - w.tracker.setFolder(box.Name, models.SyncFolderCursor{}) continue } + var selected *imap.Selected + if condStore && !stats.aborted { + view, err := client.SelectForSyncState(box.Name) + if err != nil { + return err + } + selected = &view + condStore = view.HighestModSeq > 0 + if view.UIDValidity != 0 && view.UIDValidity != box.UIDValidity { + caughtUp = false + continue + } + } changed := imapFolderChanged(befBox, box, condStore) // A pass cut short by the search cap left rows unexamined, so the // next pass looks again whether or not the count moved. @@ -128,12 +148,13 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { var view imap.Selected if changed && !stats.aborted { w.setWalking(box) - done, sel, ids, err := w.imapIncremental(ctx, box, befBox, condStore, stats) + done, sel, ids, err := w.imapIncremental(ctx, box, befBox, condStore, stats, selected) if err != nil { return err } fullyProcessed = done view = sel + condStore = condStore && view.HighestModSeq > 0 touched = ids } else if changed { // The pass was aborted before this folder; hold its cursor too. @@ -239,12 +260,15 @@ outer: } if !stats.aborted { + if err := w.imapRecoverFolders(ctx, folders, stats); err != nil { + return err + } if err := w.imapBackfill(ctx, folders, stats); err != nil { return err } } - complete = caughtUp + complete = caughtUp && !w.imapRecoveryPending(folders) return nil } @@ -461,7 +485,7 @@ const imapSkipSearchesPerPass = 50 // AND flag changes; without it only arrivals are visible here, and flag // changes are picked up by the periodic scan in imapIncremental. func imapFolderChanged(before, now *models.Mailbox, condStore bool) bool { - if condStore { + if condStore && before.HighestModSeq > 0 && now.HighestModSeq > 0 { return before.HighestModSeq != now.HighestModSeq } return before.UIDNext != now.UIDNext @@ -473,11 +497,17 @@ func imapFolderChanged(before, now *models.Mailbox, condStore bool) bool { // folder's cursor advance, the selected view the search ran against, which is // where it advances to, and the Message-IDs it fetched so the drafts // reconciliation can tell a re-appended draft from an expunged one. -func (w *WMail) imapIncremental(ctx context.Context, box, before *models.Mailbox, condStore bool, stats *tickStats) (bool, imap.Selected, map[string]struct{}, *errx.MailError) { +func (w *WMail) imapIncremental(ctx context.Context, box, before *models.Mailbox, condStore bool, stats *tickStats, selected *imap.Selected) (bool, imap.Selected, map[string]struct{}, *errx.MailError) { client := w.SmtpImapData.ImapClient - view, err := client.SelectForSyncState(box.Name) - if err != nil { - return false, view, nil, err + var view imap.Selected + var err *errx.MailError + if selected != nil { + view = *selected + } else { + view, err = client.SelectForSyncState(box.Name) + if err != nil { + return false, view, nil, err + } } // The listing and this view name different generations, so the search // would answer about UIDs the cursor does not; the next pass re-baselines. @@ -488,7 +518,7 @@ func (w *WMail) imapIncremental(ctx context.Context, box, before *models.Mailbox return true, view, nil, nil } var uids []goimap.UID - if condStore { + if condStore && before.HighestModSeq > 0 && view.HighestModSeq > 0 { uids, err = client.SearchChangedSince(before.HighestModSeq) } else { uids, err = client.SearchNewSince(before.UIDNext) @@ -515,7 +545,7 @@ func (w *WMail) imapIncremental(ctx context.Context, box, before *models.Mailbox if err != nil { return false, view, touched, err } - done, err := w.imapApply(ctx, fetched, false, stats) + done, err := w.imapApply(ctx, fetched, false, stats, nil) if err != nil { return false, view, touched, err } @@ -614,7 +644,7 @@ func (w *WMail) imapReconcileDrafts(ctx context.Context, box *models.Mailbox, to // unknown ones are stored if their lane admits them. backfill selects the // backfill lane and skips flood accounting. Returns whether every unknown // message in the batch was stored. -func (w *WMail) imapApply(ctx context.Context, fetched []*imap.Fetched, backfill bool, stats *tickStats) (bool, *errx.MailError) { +func (w *WMail) imapApply(ctx context.Context, fetched []*imap.Fetched, backfill bool, stats *tickStats, recovery *imapFolderRecovery) (bool, *errx.MailError) { sort.Slice(fetched, func(i, j int) bool { return fetched[i].Email.UID > fetched[j].Email.UID }) var fresh []*imap.Fetched @@ -667,7 +697,7 @@ func (w *WMail) imapApply(ctx context.Context, fetched []*imap.Fetched, backfill if stats.aborted { return false, nil } - if backfill && w.tracker.state.BackfillSynced >= policy.BackfillMessages { + if backfill && w.imapImportCount(recovery) >= policy.BackfillMessages { return false, nil } if !w.admit(ctx, w.laneOf(ctx, f.Email.MessageID, f.Email, backfill), stats) { @@ -677,15 +707,24 @@ func (w *WMail) imapApply(ctx context.Context, fetched []*imap.Fetched, backfill } continue } - w.SmtpImapData.ImapClient.FetchBody(f) + if err := w.SmtpImapData.ImapClient.FetchBody(f); err != nil { + if err.Code == errx.MailErrorCodeNotFound { + continue + } + return false, err + } if err := w.imapStore(ctx, f.Email); err != nil { return false, w.controlPlaneError(err, stats) } w.laneCache.forget(f.Email.MessageID) if backfill { - w.tracker.state.BackfillSynced++ - w.tracker.mark() - w.tracker.setFolder(w.SmtpImapData.folderPath, models.SyncFolderCursor{UID: f.Email.UID}) + if recovery != nil { + recovery.Synced++ + } else { + w.tracker.state.BackfillSynced++ + w.tracker.mark() + } + w.imapImportCursor(w.SmtpImapData.folderPath, f.Email.UID, false, recovery) } } return all, nil @@ -870,7 +909,7 @@ func (w *WMail) imapBackfill(ctx context.Context, folders []models.Mailbox, stat if err != nil { return err } - done, err := w.imapApply(ctx, fetched, true, stats) + done, err := w.imapApply(ctx, fetched, true, stats, nil) if err != nil { return err } diff --git a/internal/app/worker/wmail/sync_imap_recovery.go b/internal/app/worker/wmail/sync_imap_recovery.go new file mode 100644 index 000000000..18eb31c05 --- /dev/null +++ b/internal/app/worker/wmail/sync_imap_recovery.go @@ -0,0 +1,146 @@ +package wmail + +import ( + "context" + "encoding/json" + "sort" + "strings" + "time" + + goimap "github.com/emersion/go-imap/v2" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +const imapRecoveryPrefix = "imap-recovery-v1:" + +// Recovery metadata lives in the existing opaque folder continuation, not a new wire field. +type imapFolderRecovery struct { + Generation uint32 `json:"generation"` + Since time.Time `json:"since"` + Synced int `json:"synced"` +} + +func imapRecoveryCursor(cur models.SyncFolderCursor) *imapFolderRecovery { + raw, ok := strings.CutPrefix(cur.Next, imapRecoveryPrefix) + if !ok { + return nil + } + var recovery imapFolderRecovery + if json.Unmarshal([]byte(raw), &recovery) != nil || recovery.Generation == 0 || recovery.Since.IsZero() || recovery.Synced < 0 { + return nil + } + return &recovery +} + +func (w *WMail) imapRecoverFolder(box *models.Mailbox) bool { + if w.tracker.state.BackfillStatus != models.SyncBackfillComplete || !imapBackfillEligible(box) { + return true + } + cur := w.tracker.folder(box.Name) + if recovery := imapRecoveryCursor(cur); recovery == nil || recovery.Generation != box.UIDValidity { + since := time.Now().Add(-time.Duration(w.gov.Policy().BackfillDays) * 24 * time.Hour) + w.imapImportCursor(box.Name, 0, false, &imapFolderRecovery{Generation: box.UIDValidity, Since: since}) + } + // Publish replay state before a live baseline can make the folder look caught up on reload. + w.tracker.flush(time.Now()) + return !w.tracker.dirty +} + +func (w *WMail) imapRecoveryPending(folders []models.Mailbox) bool { + for _, box := range folders { + cur := w.tracker.folder(box.Name) + if !cur.Done && imapRecoveryCursor(cur) != nil && imapBackfillEligible(&box) { + return true + } + } + return false +} + +func (w *WMail) imapImportCount(recovery *imapFolderRecovery) int { + if recovery != nil { + return recovery.Synced + } + return w.tracker.state.BackfillSynced +} + +func (w *WMail) imapImportCursor(folder string, uid uint32, done bool, recovery *imapFolderRecovery) { + cur := models.SyncFolderCursor{UID: uid, Done: done} + if recovery != nil { + raw, _ := json.Marshal(recovery) + cur.Next = imapRecoveryPrefix + string(raw) + } + w.tracker.setFolder(folder, cur) +} + +// Completed account history does not suppress bounded recovery of a new folder or UID generation. +func (w *WMail) imapRecoverFolders(ctx context.Context, folders []models.Mailbox, stats *tickStats) *errx.MailError { + client := w.SmtpImapData.ImapClient + for i := range folders { + box := &folders[i] + cur := w.tracker.folder(box.Name) + recovery := imapRecoveryCursor(cur) + if recovery == nil || !imapBackfillEligible(box) { + continue + } + if recovery.Generation != box.UIDValidity { + if !w.imapRecoverFolder(box) { + stats.aborted = true + return nil + } + cur = w.tracker.folder(box.Name) + recovery = imapRecoveryCursor(cur) + if recovery == nil { + continue + } + } + if cur.Done { + continue + } + if ctx.Err() != nil || stats.aborted || stats.laneDenied(LaneBackfill) { + return nil + } + if recovery.Synced >= w.gov.Policy().BackfillMessages { + w.imapImportCursor(box.Name, cur.UID, true, recovery) + continue + } + w.setWalking(box) + view, err := client.SelectForSyncState(box.Name) + if err != nil { + return err + } + if view.UIDValidity != 0 && view.UIDValidity != recovery.Generation { + return errx.ErrMailResourceNotFound + } + uids, err := client.SearchSince(recovery.Since) + if err != nil { + return err + } + remaining := uids[:0:0] + for _, uid := range uids { + if cur.UID == 0 || uid < goimap.UID(cur.UID) { + remaining = append(remaining, uid) + } + } + sort.Slice(remaining, func(i, j int) bool { return remaining[i] > remaining[j] }) + for lo := 0; lo < len(remaining); lo += config.ImapFetchBatchSize { + hi := min(lo+config.ImapFetchBatchSize, len(remaining)) + fetched, err := client.FetchEnvelopes(ctx, remaining[lo:hi]) + if err != nil { + return err + } + done, err := w.imapApply(ctx, fetched, true, stats, recovery) + if err != nil { + return err + } + if !done || ctx.Err() != nil { + return nil + } + cur.UID = uint32(remaining[hi-1]) + w.imapImportCursor(box.Name, cur.UID, false, recovery) + } + w.imapImportCursor(box.Name, cur.UID, true, recovery) + } + return nil +} diff --git a/internal/app/worker/wmail/sync_imap_recovery_test.go b/internal/app/worker/wmail/sync_imap_recovery_test.go new file mode 100644 index 000000000..9a91d0142 --- /dev/null +++ b/internal/app/worker/wmail/sync_imap_recovery_test.go @@ -0,0 +1,229 @@ +package wmail + +import ( + "encoding/json" + "errors" + "testing" + "time" + + goimap "github.com/emersion/go-imap/v2" + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/client/smtpimap/imap" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +func imapReload(t *testing.T, w *WMail, conn ImapConn, budget syncBudget) (*WMail, *[]captured) { + t.Helper() + // Use the existing JSON wire shape, not shared in-memory state, to model restart. + raw, err := json.Marshal(w.tracker.state) + if err != nil { + t.Fatal(err) + } + var state models.SyncState + if err := json.Unmarshal(raw, &state); err != nil { + t.Fatal(err) + } + saved := *w.SmtpImapData.Mailboxes[0] + next, events := newIMAPTestMail(conn, budget, &saved) + next.EmailMessageMapRepository = w.EmailMessageMapRepository + next.tracker = newSyncTracker(&state, func(models.SyncState) error { return nil }) + return next, events +} + +func TestImapCompletedFolderRecoveryResumesAfterDeferralAndReload(t *testing.T) { + for _, generation := range []bool{false, true} { + t.Run(map[bool]string{false: "new populated folder", true: "new generation"}[generation], func(t *testing.T) { + box := models.Mailbox{Name: "Archive", UIDValidity: 9, UIDNext: 4, HighestModSeq: 100} + conn := &fakeImapConn{folders: []models.Mailbox{box}, all: []goimap.UID{1, 2, 3}} + saved := &models.Mailbox{Name: "Archive", UIDValidity: 8, UIDNext: 90, HighestModSeq: 100} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 1}, saved) + if !generation { + w.SmtpImapData.Mailboxes = nil + } + maps := &recoveryMessageMap{data: map[string]repository.EmailMessageData{}} + knownID := uuid.NewString() + maps.data["<2@fake.test>"] = repository.EmailMessageData{ID: knownID, MessageID: "<2@fake.test>"} + w.EmailMessageMapRepository = maps + w.tracker.setFolder("Archive", models.SyncFolderCursor{UID: 80, Done: true}) + oldSuccess := time.Now().Add(-time.Hour) + w.tracker.state.LastSyncedAt = &oldSuccess + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + cur := w.tracker.folder("Archive") + recovery := imapRecoveryCursor(cur) + if cur.Done || cur.UID != 3 || recovery == nil || recovery.Generation != 9 || recovery.Synced != 1 { + t.Fatalf("deferred recovery = %+v / %+v", cur, recovery) + } + if len(mailboxEvents(*events, models.JobEventTypeNewEmail)) != 1 { + t.Fatal("newest recovery arrival missing") + } + if !w.tracker.state.LastSyncedAt.Equal(oldSuccess) { + t.Fatal("deferred recovery refreshed success") + } + cutoff := recovery.Since + next, resumed := imapReload(t, w, conn, &fixedBudget{allow: 10}) + if err := next.Sync(t.Context()); err != nil { + t.Fatal(err) + } + cur = next.tracker.folder("Archive") + if !cur.Done || cur.UID != 1 || !imapRecoveryCursor(cur).Since.Equal(cutoff) { + t.Fatalf("restarted recovery = %+v", cur) + } + if len(maps.data) != 3 || len(mailboxEvents(*resumed, models.JobEventTypeNewEmail)) != 1 || maps.data["<2@fake.test>"].ID != knownID { + t.Fatal("recovery lost or duplicated arrivals") + } + if next.tracker.state.BackfillStatus != models.SyncBackfillComplete || next.tracker.state.BackfillSynced != 0 { + t.Fatal("global history reset") + } + before := conn.searches + again, later := imapReload(t, next, conn, &fixedBudget{allow: 10}) + if err := again.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if conn.searches != before || len(mailboxEvents(*later, models.JobEventTypeNewEmail)) != 0 { + t.Fatal("completed recovery reimported on restart") + } + }) + } +} + +func TestImapRecoveryDoesNotBaselineBeforeReplayStateRelays(t *testing.T) { + conn := &fakeImapConn{folders: []models.Mailbox{{Name: "Archive", UIDValidity: 9, UIDNext: 3, HighestModSeq: 100}}, all: []goimap.UID{1, 2}} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, &models.Mailbox{Name: "Archive", UIDValidity: 8}) + w.tracker.emit = func(models.SyncState) error { return errors.New("state unavailable") } + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if w.SmtpImapData.Mailboxes[0].UIDValidity != 8 || len(mailboxEvents(*events, models.JobEventTypeMailboxUpdate)) != 0 || conn.fetches != 0 { + t.Fatal("baseline outran replay checkpoint") + } + w.tracker.emit = func(models.SyncState) error { return nil } + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if !w.tracker.folder("Archive").Done || len(mailboxEvents(*events, models.JobEventTypeNewEmail)) != 2 { + t.Fatal("checkpoint retry did not recover folder") + } +} + +func TestImapBodyFailurePinsLiveAndRecoveryUntilReload(t *testing.T) { + for _, recovery := range []bool{false, true} { + t.Run(map[bool]string{false: "live", true: "recovery"}[recovery], func(t *testing.T) { + box := models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 2, HighestModSeq: 101} + conn := &fakeImapConn{folders: []models.Mailbox{box}, changed: []goimap.UID{1}, all: []goimap.UID{1}, bodyErr: errx.ErrMailServerUnreachable} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, &models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 1, HighestModSeq: 100}) + if recovery { + w.SmtpImapData.Mailboxes[0].UIDValidity = 6 + } + maps := &recoveryMessageMap{data: map[string]repository.EmailMessageData{}} + w.EmailMessageMapRepository = maps + if err := w.Sync(t.Context()); err == nil { + t.Fatal("body failure swallowed") + } + if len(maps.data) != 0 || len(mailboxEvents(*events, models.JobEventTypeNewEmail)) != 0 || w.tracker.state.BackfillSynced != 0 || w.tracker.folder("INBOX").UID != 0 || w.tracker.folder("INBOX").Done { + t.Fatal("body failure imported partial message or advanced recovery") + } + if !recovery && w.SmtpImapData.Mailboxes[0].UIDNext != 1 { + t.Fatal("body failure advanced live cursor") + } + conn.bodyErr = nil + next, resumed := imapReload(t, w, conn, &fixedBudget{allow: 10}) + if err := next.Sync(t.Context()); err != nil { + t.Fatal(err) + } + arrivals := mailboxEvents(*resumed, models.JobEventTypeNewEmail) + if len(arrivals) != 1 || len(maps.data) != 1 { + t.Fatal("hydration was not retried exactly once") + } + if arrivals[0].body.(*models.JobEventNewEmail).Message.BodyText != "complete body" { + t.Fatal("arrival did not contain hydrated body") + } + if next.SmtpImapData.Mailboxes[0].UIDNext != 2 { + t.Fatal("live cursor did not advance after hydration") + } + }) + } +} + +func TestImapNomodseqUsesUIDNextAndPeriodicFlags(t *testing.T) { + for _, listedModseq := range []uint64{0, 100} { + t.Run(map[uint64]string{0: "status NOMODSEQ", 100: "select NOMODSEQ"}[listedModseq], func(t *testing.T) { + conn := &fakeImapConn{folders: []models.Mailbox{{Name: "INBOX", UIDValidity: 7, UIDNext: 3, HighestModSeq: listedModseq}}, changed: []goimap.UID{2}, view: &imap.Selected{UIDNext: 3}, flags: map[uint32]imap.FlagState{1: {MessageID: ""}}} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, &models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 2, HighestModSeq: listedModseq}) + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if len(mailboxEvents(*events, models.JobEventTypeNewEmail)) != 1 || conn.flagScans != 1 || conn.uidSearches != 1 || conn.modseqSearches != 0 { + t.Fatal("NOMODSEQ ignored arrival or fallback flags") + } + w.EmailMessageMapRepository = knownMessageMap{id: uuid.NewString()} + conn.flags[1] = imap.FlagState{MessageID: "", Flags: []string{`\Seen`}} + w.flagScan["INBOX"].at = time.Now().Add(-2 * config.ImapFlagScanInterval) + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if len(mailboxEvents(*events, models.JobEventTypeEmailUpdate)) != 1 { + t.Fatal("NOMODSEQ dropped periodic read update") + } + }) + } +} + +func TestImapBodyFailurePinsInitialHistoryUntilReload(t *testing.T) { + box := models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 2, HighestModSeq: 100, Messages: 1} + conn := &fakeImapConn{folders: []models.Mailbox{box}, changed: []goimap.UID{1}, all: []goimap.UID{1}, bodyErr: errx.ErrMailServerUnreachable} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, &box) + w.tracker = newSyncTracker(nil, func(models.SyncState) error { return nil }) + maps := &recoveryMessageMap{data: map[string]repository.EmailMessageData{}} + w.EmailMessageMapRepository = maps + if err := w.Sync(t.Context()); err == nil { + t.Fatal("history body failure swallowed") + } + if len(maps.data) != 0 || len(mailboxEvents(*events, models.JobEventTypeNewEmail)) != 0 || w.tracker.state.BackfillSynced != 0 || w.tracker.folder("INBOX").UID != 0 || w.tracker.folder("INBOX").Done { + t.Fatal("history advanced after body failure") + } + conn.bodyErr = nil + next, resumed := imapReload(t, w, conn, &fixedBudget{allow: 10}) + if err := next.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if len(maps.data) != 1 || len(mailboxEvents(*resumed, models.JobEventTypeNewEmail)) != 1 || next.tracker.state.BackfillSynced != 1 || !next.tracker.folder("INBOX").Done { + t.Fatal("history hydration did not resume") + } +} + +func TestImapFlagScanRetriesBothChangesAfterPublicationAbort(t *testing.T) { + conn := &fakeImapConn{folders: []models.Mailbox{{Name: "INBOX", UIDValidity: 7, UIDNext: 3}}, noCondStore: true, flags: map[uint32]imap.FlagState{1: {MessageID: ""}, 2: {MessageID: ""}}} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, &models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 3}) + w.EmailMessageMapRepository = knownMessageMap{id: uuid.NewString()} + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + for uid, st := range conn.flags { + st.Flags = []string{`\Seen`} + conn.flags[uid] = st + } + emit := w.onEvent + w.onEvent = func(kind models.JobEventType, body any) error { + if kind == models.JobEventTypeEmailUpdate { + return errors.New("publish unavailable") + } + return emit(kind, body) + } + w.flagScan["INBOX"].at = time.Now().Add(-2 * config.ImapFlagScanInterval) + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + w.onEvent = emit + w.flagScan["INBOX"].at = time.Now().Add(-2 * config.ImapFlagScanInterval) + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + if len(mailboxEvents(*events, models.JobEventTypeEmailUpdate)) != 2 { + t.Fatal("failed or unvisited flag changes were absorbed into baseline") + } +} diff --git a/internal/app/worker/wmail/sync_imap_test.go b/internal/app/worker/wmail/sync_imap_test.go index 752986310..5cc493ee7 100644 --- a/internal/app/worker/wmail/sync_imap_test.go +++ b/internal/app/worker/wmail/sync_imap_test.go @@ -43,7 +43,11 @@ type fakeImapConn struct { failFinds bool // view, when set, is the cursors SELECT reports in place of the listing's: // a server whose selected view lags or leads its STATUS. - view *imap.Selected + view *imap.Selected + searches int + bodyErr *errx.MailError + modseqSearches int + uidSearches int } // Folders hands out a copy, like a real listing: the pass filters the slice @@ -61,6 +65,7 @@ func (c *fakeImapConn) FolderConflicts() int { return c.conflicts } func (c *fakeImapConn) HasCondStore() bool { return !c.noCondStore } func (c *fakeImapConn) SearchNewSince(uidNext uint32) ([]goimap.UID, *errx.MailError) { + c.uidSearches++ var out []goimap.UID for _, uid := range c.changed { if uint32(uid) >= uidNext { @@ -115,6 +120,7 @@ func (c *fakeImapConn) SelectForSyncGen(name string) (uint32, uint32, *errx.Mail } func (c *fakeImapConn) SearchChangedSince(uint64) ([]goimap.UID, *errx.MailError) { + c.modseqSearches++ return append([]goimap.UID(nil), c.changed...), nil } @@ -122,6 +128,11 @@ func (c *fakeImapConn) SearchAll() ([]goimap.UID, *errx.MailError) { return append([]goimap.UID(nil), c.all...), nil } +func (c *fakeImapConn) SearchSince(time.Time) ([]goimap.UID, *errx.MailError) { + c.searches++ + return append([]goimap.UID(nil), c.all...), nil +} + func (c *fakeImapConn) FetchEnvelopes(_ context.Context, uids []goimap.UID) ([]*imap.Fetched, *errx.MailError) { c.fetches++ out := make([]*imap.Fetched, 0, len(uids)) @@ -135,7 +146,13 @@ func (c *fakeImapConn) FetchEnvelopes(_ context.Context, uids []goimap.UID) ([]* return out, nil } -func (c *fakeImapConn) FetchBody(*imap.Fetched) {} +func (c *fakeImapConn) FetchBody(f *imap.Fetched) *errx.MailError { + if c.bodyErr != nil { + return c.bodyErr + } + f.Email.BodyPlain = "complete body" + return nil +} // fixedBudget is a syncBudget that admits a fixed number of messages and then // denies on the daily window, which is how a real governor answers once the @@ -339,6 +356,17 @@ func (c *backfillImapConn) FolderConflicts() int { retur func (c *backfillImapConn) HasCondStore() bool { return true } func (c *backfillImapConn) ReleaseMailbox() {} +func (c *backfillImapConn) SelectForSyncState(folder string) (imap.Selected, *errx.MailError) { + count, err := c.SelectForSync(folder) + view := imap.Selected{Count: count} + for _, box := range c.folders { + if box.Name == folder { + view.UIDValidity, view.UIDNext, view.HighestModSeq = box.UIDValidity, box.UIDNext, box.HighestModSeq + } + } + return view, err +} + func (c *backfillImapConn) SelectForSync(name string) (uint32, *errx.MailError) { c.selected = name return uint32(len(c.uids[name])), nil @@ -375,7 +403,7 @@ func (c *backfillImapConn) FetchEnvelopes(_ context.Context, uids []goimap.UID) return out, nil } -func (c *backfillImapConn) FetchBody(*imap.Fetched) {} +func (c *backfillImapConn) FetchBody(*imap.Fetched) *errx.MailError { return nil } // The Graph defect's shape, checked on the IMAP import: a folder whose search // fails holds its cursor and is walked again on the next pass. Nothing about a diff --git a/internal/client/smtpimap/imap/body_read.go b/internal/client/smtpimap/imap/body_read.go index 3d198416b..eb8b75640 100644 --- a/internal/client/smtpimap/imap/body_read.go +++ b/internal/client/smtpimap/imap/body_read.go @@ -3,6 +3,7 @@ package imap import ( "bytes" "encoding/base64" + "errors" "io" "mime/quotedprintable" "strconv" @@ -35,14 +36,14 @@ const maxTextParts = 5 // bytes back as the HTML body too, so plain text was rendered as markup: line // breaks collapsed, "&" showed as an entity, and anything inside angle // brackets vanished. -func fetchTextParts(c *imapclient.Client, uid imap.UID, bs imap.BodyStructure) (plain, html string) { +func fetchTextParts(c *imapclient.Client, uid imap.UID, bs imap.BodyStructure) (plain, html string, err error) { if bs == nil { - return "", "" + return "", "", nil } wanted := selectTextParts(bs) if len(wanted) == 0 { - return "", "" + return "", "", nil } sections := make([]*imap.FetchItemBodySection, 0, len(wanted)) @@ -56,12 +57,15 @@ func fetchTextParts(c *imapclient.Client, uid imap.UID, bs imap.BodyStructure) ( }) } - raw := fetchSections(c, uid, sections) + raw, err := fetchSections(c, uid, sections) + if err != nil { + return "", "", err + } var plainParts, htmlParts []string for _, p := range wanted { body, ok := raw[partKey(p.path)] if !ok { - continue + return "", "", errors.New("imap: required body section missing") } decoded := decodePart(body, p.encoding, p.charset) if p.html { @@ -72,7 +76,7 @@ func fetchTextParts(c *imapclient.Client, uid imap.UID, bs imap.BodyStructure) ( } // Sibling inline parts (a body followed by a disclaimer, say) are pieces of // one message, so they are joined rather than one replacing the other. - return strings.Join(plainParts, "\n\n"), strings.Join(htmlParts, "") + return strings.Join(plainParts, "\n\n"), strings.Join(htmlParts, ""), nil } // reportPartTypes are the machine-readable parts of a multipart/report. They @@ -159,16 +163,20 @@ func selectTextParts(bs imap.BodyStructure) []textPart { // fetchSections runs one FETCH for the given sections and returns each part's // raw (still transfer-encoded) bytes, keyed by part path. -func fetchSections(c *imapclient.Client, uid imap.UID, sections []*imap.FetchItemBodySection) map[string][]byte { +var errBodyMessageGone = errors.New("imap: body message no longer exists") + +func fetchSections(c *imapclient.Client, uid imap.UID, sections []*imap.FetchItemBodySection) (map[string][]byte, error) { out := make(map[string][]byte, len(sections)) cmd := c.Fetch(imap.UIDSetNum(uid), &imap.FetchOptions{ UID: true, BodySection: sections, }) - defer cmd.Close() + seen := false + var readErr error for msg := cmd.Next(); msg != nil; msg = cmd.Next() { + seen = true for item := msg.Next(); item != nil; item = msg.Next() { v, ok := item.(imapclient.FetchItemDataBodySection) if !ok || v.Literal == nil || v.Section == nil { @@ -178,13 +186,29 @@ func fetchSections(c *imapclient.Client, uid imap.UID, sections []*imap.FetchIte // Belt and braces alongside the server-side Partial: a server that // ignores the partial range must not be able to stream us an // unbounded body. - if _, err := io.Copy(buf, io.LimitReader(v.Literal, int64(config.MaxEmailBodySize))); err != nil { + n, err := io.Copy(buf, io.LimitReader(v.Literal, int64(config.MaxEmailBodySize))) + if err == nil && n < min(v.Literal.Size(), int64(config.MaxEmailBodySize)) { + err = io.ErrUnexpectedEOF + } + if err != nil { + // Stop the decoder before go-imap tries discarding a short literal a second time. + _ = c.Close() + readErr = err continue } out[partKey(v.Section.Part)] = buf.Bytes() } } - return out + if err := cmd.Close(); err != nil { + return nil, err + } + if readErr != nil { + return nil, readErr + } + if !seen { + return nil, errBodyMessageGone + } + return out, nil } // partKey renders an IMAP part path ("1", "1.2") so a fetched section can be diff --git a/internal/client/smtpimap/imap/client.go b/internal/client/smtpimap/imap/client.go index ec9ea19e9..e5ea1db29 100644 --- a/internal/client/smtpimap/imap/client.go +++ b/internal/client/smtpimap/imap/client.go @@ -35,6 +35,8 @@ var headerFetchFields = append([]string{config.WarmupVerifyHeader}, config.Inbou var evidenceFetchFields = append(append([]string{}, headerFetchFields...), "Authentication-Results", "From", "Return-Path", "DKIM-Signature", "List-Unsubscribe", "List-Unsubscribe-Post") type Client struct { + // Context optionally binds connections and setup to the installed mailbox lifetime. + Context context.Context Email string AuthType models.AuthType Credentials *models.Service @@ -79,6 +81,12 @@ type Client struct { // like client itself. conn *idleConn + setupMu sync.Mutex + setupCancel context.CancelFunc + transport net.Conn + lifetimeStop func() bool + closed atomic.Bool + // IdleTimeout bounds how long a command waits for the server to say // anything before the session is declared dead. Zero means // config.ImapCommandIdleTimeout. @@ -130,6 +138,11 @@ func (c *Client) HasCondStore() bool { return c.condStore.Load() } +func (c *Client) selectedCondStore() bool { + sel := c.selection.Load() + return c.condStore.Load() && sel != nil && sel.modSeq > 0 +} + // ensureConnected re-dials after the server has dropped the session. go-imap // parks a dead client in the Logout state and fails every later command with // net.ErrClosed; nothing re-dialed, so one drop (Gmail closes sessions after a @@ -139,6 +152,9 @@ func (c *Client) HasCondStore() bool { func (c *Client) ensureConnected() *errx.MailError { c.lifecycle.Lock() defer c.lifecycle.Unlock() + if c.closed.Load() || c.Context != nil && c.Context.Err() != nil { + return errx.ErrMailServerUnreachable + } // Only a session that got past auth is worth keeping: a failed Login // leaves go-imap in NotAuthenticated, which is just as unusable as Logout. @@ -160,6 +176,18 @@ func (c *Client) Connect() *errx.MailError { // connectLocked dials and authenticates a fresh session. lifecycle must be // held for writing. func (c *Client) connectLocked() *errx.MailError { + if c.closed.Load() || c.Context != nil && c.Context.Err() != nil { + return errx.ErrMailServerUnreachable + } + c.setupMu.Lock() + if c.lifetimeStop != nil { + c.lifetimeStop() + } + if c.transport != nil { + _ = c.transport.Close() + c.transport = nil + } + c.setupMu.Unlock() var addr, host, security string var port int switch c.AuthType { @@ -193,7 +221,26 @@ func (c *Client) connectLocked() *errx.MailError { if resolved == models.MailSecurityNone && !models.CleartextMailAllowed(host) { return errx.ErrMailInsecureRemoteHost } - raw, err := netbind.Dialer(c.BindIP).DialContext(context.Background(), "tcp", addr) + lifetime := c.Context + if lifetime == nil { + lifetime = context.Background() + } + setupCtx, cancel := context.WithTimeout(lifetime, timeout) + c.setupMu.Lock() + if c.closed.Load() { + c.setupMu.Unlock() + cancel() + return errx.ErrMailServerUnreachable + } + c.setupCancel = cancel + c.setupMu.Unlock() + defer func() { + c.setupMu.Lock() + c.setupCancel = nil + c.setupMu.Unlock() + cancel() + }() + raw, err := netbind.Dialer(c.BindIP).DialContext(setupCtx, "tcp", addr) if err != nil { return errx.ErrMailServerUnreachable } @@ -201,7 +248,21 @@ func (c *Client) connectLocked() *errx.MailError { _ = raw.Close() return errx.ErrMailInsecureRemoteHost } + c.setupMu.Lock() + if c.closed.Load() { + c.setupMu.Unlock() + _ = raw.Close() + return errx.ErrMailServerUnreachable + } + c.transport = raw + c.lifetimeStop = context.AfterFunc(lifetime, func() { _ = raw.Close() }) + c.setupMu.Unlock() conn := &idleConn{Conn: raw, timeout: timeout} + // Library greeting/STARTTLS waits precede client installation; close the pending socket too. + stopSetup := context.AfterFunc(setupCtx, func() { _ = raw.Close() }) + defer stopSetup() + done := conn.arm() + defer done() var client *imapclient.Client switch { @@ -216,9 +277,7 @@ func (c *Client) connectLocked() *errx.MailError { } default: tconn := tls.Client(conn, tlsConf) - hctx, cancel := context.WithTimeout(context.Background(), timeout) - err = tconn.HandshakeContext(hctx) - cancel() + err = tconn.HandshakeContext(setupCtx) if err != nil { _ = tconn.Close() return errx.ErrMailServerUnreachable @@ -231,9 +290,6 @@ func (c *Client) connectLocked() *errx.MailError { c.selected.Store(false) c.selection.Store(nil) c.condStore.Store(false) - done := conn.arm() - defer done() - var xerr *errx.MailError switch c.AuthType { @@ -254,17 +310,30 @@ func (c *Client) connectLocked() *errx.MailError { // check must run post-auth. Without it (Outlook.com, Microsoft 365 over // IMAP, Yahoo, many hosted servers) the sync loop keys on UIDNEXT instead. c.condStore.Store(c.client.Caps().Has(imap.CapCondStore)) + if !stopSetup() || setupCtx.Err() != nil { + _ = client.Close() + return errx.ErrMailServerUnreachable + } return nil } func (c *Client) Close() error { - c.lifecycle.RLock() - defer c.lifecycle.RUnlock() - if c.client == nil { + if !c.closed.CompareAndSwap(false, true) { return nil } - return c.client.Close() + c.setupMu.Lock() + defer c.setupMu.Unlock() + if c.setupCancel != nil { + c.setupCancel() + } + if c.lifetimeStop != nil { + c.lifetimeStop() + } + if c.transport == nil { + return nil + } + return c.transport.Close() } func (c *Client) plainAuth() *errx.MailError { @@ -312,7 +381,7 @@ func (c *Client) selectMailbox(mailbox string, opts *imap.SelectOptions) (*imap. data, err := c.client.Select(mailbox, opts).Wait() c.selected.Store(err == nil) if err == nil { - c.selection.Store(&selection{name: mailbox, uidValidity: data.UIDValidity, count: data.NumMessages}) + c.selection.Store(&selection{name: mailbox, uidValidity: data.UIDValidity, count: data.NumMessages, modSeq: data.HighestModSeq}) } return data, err } @@ -649,7 +718,7 @@ func (c *Client) FetchEnvelopes(ctx context.Context, uids []imap.UID) ([]*Fetche // never advertised it for one is a malformed fetch-att, and a strict // parser answers BAD and syncs nothing: issue #405, where IONOS said // `BAD expected fetch-att instead of "MODSEQ BODY.P"` on every pass. - ModSeq: c.condStore.Load(), + ModSeq: c.selectedCondStore(), InternalDate: true, RFC822Size: true, BodySection: []*imap.FetchItemBodySection{{ @@ -720,9 +789,9 @@ func (c *Client) FetchEnvelopes(ctx context.Context, uids []imap.UID) ([]*Fetche // the FetchEnvelopes command that produced it is closed: a nested FETCH on // the same connection blocks until the outer one finishes, and the outer one // cannot finish while we wait, which deadlocks the sync on the first message. -func (c *Client) FetchBody(f *Fetched) { +func (c *Client) FetchBody(f *Fetched) *errx.MailError { if f == nil || f.Email == nil { - return + return nil } c.mu.Lock() defer c.mu.Unlock() @@ -730,10 +799,20 @@ func (c *Client) FetchBody(f *Fetched) { defer c.lifecycle.RUnlock() defer c.begin()() if err := c.resumeSyncLocked(); err != nil { - log.Debug().Err(err).Uint32("uid", uint32(f.uid)).Msg("imap: could not re-open the folder to read a body") - return + return errx.ErrMailServerUnreachable } - f.Email.BodyPlain, f.Email.BodyHTML = fetchTextParts(c.client, f.uid, f.body) + plain, html, err := fetchTextParts(c.client, f.uid, f.body) + if errors.Is(err, errBodyMessageGone) { + return errx.ErrMailResourceNotFound + } + if err != nil { + if mailErr := c.handleError(err); mailErr.Code != errx.MailErrorCodeNotFound { + return mailErr + } + return errx.ErrMailServerUnreachable + } + f.Email.BodyPlain, f.Email.BodyHTML = plain, html + return nil } // parseHeaderFlags reads a HEADER.FIELDS literal and renders the fetched diff --git a/internal/client/smtpimap/imap/fetch_caps_test.go b/internal/client/smtpimap/imap/fetch_caps_test.go index e92cfb9d0..0fbe15b6c 100644 --- a/internal/client/smtpimap/imap/fetch_caps_test.go +++ b/internal/client/smtpimap/imap/fetch_caps_test.go @@ -87,6 +87,10 @@ func TestFetchEnvelopesAsksModSeqWithCondStore(t *testing.T) { t.Fatalf("SelectForSync: %v", err) } c.condStore.Store(true) + selected := *c.selection.Load() + selected.modSeq = 1 + c.selection.Store(&selected) + c.syncView = &selected // The result is deliberately ignored: this server refuses MODSEQ when // CONDSTORE was not enabled, which is #405's failure reproduced in diff --git a/internal/client/smtpimap/imap/search_date.go b/internal/client/smtpimap/imap/search_date.go index 45b6efc73..468244a0d 100644 --- a/internal/client/smtpimap/imap/search_date.go +++ b/internal/client/smtpimap/imap/search_date.go @@ -22,6 +22,7 @@ type selection struct { name string uidValidity uint32 count uint32 + modSeq uint64 } // dateScan is the INTERNALDATE read of one folder for one window. UIDs below diff --git a/internal/client/smtpimap/imap/sync_repair_wire_test.go b/internal/client/smtpimap/imap/sync_repair_wire_test.go new file mode 100644 index 000000000..1e8f53174 --- /dev/null +++ b/internal/client/smtpimap/imap/sync_repair_wire_test.go @@ -0,0 +1,490 @@ +package imap + +import ( + "bufio" + "context" + "crypto/tls" + "crypto/x509" + "fmt" + "io" + "net" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + goimap "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func repairWireServer(t *testing.T, serve func(net.Conn)) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func() { defer func() { _ = conn.Close() }(); serve(conn) }() + } + }() + return ln.Addr().String() +} + +// This fixture advertises global CONDSTORE but selects a NOMODSEQ folder. +func bodyRepairFixture(t *testing.T, bodyReply func(net.Conn, string) bool) (*Client, *wireLog) { + t.Helper() + wire := &wireLog{} + addr := repairWireServer(t, func(conn net.Conn) { + _, _ = io.WriteString(conn, "* OK [CAPABILITY IMAP4rev1 CONDSTORE] ready\r\n") + r := bufio.NewReader(conn) + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + wire.add(strings.TrimSpace(line)) + fields := strings.Fields(line) + if len(fields) < 2 { + return + } + tag := fields[0] + switch strings.ToUpper(fields[1]) { + case "LOGIN": + _, _ = fmt.Fprintf(conn, "%s OK [CAPABILITY IMAP4rev1 CONDSTORE] authenticated\r\n", tag) + case "CAPABILITY": + _, _ = fmt.Fprintf(conn, "* CAPABILITY IMAP4rev1 CONDSTORE\r\n%s OK capability\r\n", tag) + case "EXAMINE": + _, _ = fmt.Fprintf(conn, "* 1 EXISTS\r\n* OK [UIDVALIDITY 7] generation\r\n* OK [UIDNEXT 2] next\r\n* OK [NOMODSEQ] no modseq\r\n%s OK [READ-ONLY] selected\r\n", tag) + case "UID": + if len(fields) > 2 && strings.EqualFold(fields[2], "SEARCH") { + _, _ = fmt.Fprintf(conn, "* SEARCH 1\r\n%s OK search\r\n", tag) + } else if strings.Contains(strings.ToUpper(line), "BODYSTRUCTURE") { + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1 FLAGS () ENVELOPE (NIL \"hello\" NIL NIL NIL NIL NIL NIL NIL \"\") BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"UTF-8\") NIL NIL \"7BIT\" 13 1))\r\n%s OK envelope\r\n", tag) + } else if bodyReply != nil { + if !bodyReply(conn, tag) { + return + } + } else { + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1 FLAGS (\\Seen) ENVELOPE (NIL NIL NIL NIL NIL NIL NIL NIL NIL \"\"))\r\n%s OK flags\r\n", tag) + } + default: + _, _ = fmt.Fprintf(conn, "%s BAD unexpected command\r\n", tag) + } + } + }) + return clientTo(t, addr), wire +} + +func TestFetchBodyWireErrorsRemainRetryable(t *testing.T) { + for _, failure := range []string{"tagged refusal", "short literal", "reset literal", "missing section", "failed completion"} { + t.Run(failure, func(t *testing.T) { + var attempts atomic.Int32 + c, _ := bodyRepairFixture(t, func(conn net.Conn, tag string) bool { + if attempts.Add(1) == 1 { + switch failure { + case "tagged refusal": + _, _ = fmt.Fprintf(conn, "%s NO [NONEXISTENT] body refused\r\n", tag) + case "short literal": + _, _ = io.WriteString(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\nshort") + return false + case "reset literal": + _, _ = io.WriteString(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\nshort") + _ = conn.(*net.TCPConn).SetLinger(0) + return false + case "missing section": + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1)\r\n%s OK completed\r\n", tag) + case "failed completion": + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\ncomplete body)\r\n%s NO incomplete\r\n", tag) + } + return true + } + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\ncomplete body)\r\n%s OK body\r\n", tag) + return true + }) + c.IdleTimeout = time.Second + if err := c.Connect(); err != nil { + t.Fatal(err) + } + if _, err := c.SelectForSync("INBOX"); err != nil { + t.Fatal(err) + } + fetched, err := c.FetchEnvelopes(t.Context(), []goimap.UID{1}) + if err != nil || len(fetched) != 1 { + t.Fatalf("envelope: %v %v", fetched, err) + } + if err := c.FetchBody(fetched[0]); err == nil || err.Code == errx.MailErrorCodeNotFound { + t.Fatalf("hydration failure treated as bodyless or gone: %v", err) + } + if fetched[0].Email.BodyPlain != "" || fetched[0].Email.BodyHTML != "" { + t.Fatal("failed body installed partial text") + } + if err := c.ensureConnected(); err != nil { + t.Fatal(err) + } + if _, err := c.SelectForSync("INBOX"); err != nil { + t.Fatal(err) + } + fetched, err = c.FetchEnvelopes(t.Context(), []goimap.UID{1}) + if err != nil || len(fetched) != 1 { + t.Fatalf("retry envelope: %v %v", fetched, err) + } + if err := c.FetchBody(fetched[0]); err != nil { + t.Fatal(err) + } + if fetched[0].Email.BodyPlain != "complete body" { + t.Fatalf("retry body = %q", fetched[0].Email.BodyPlain) + } + }) + } +} + +func TestFetchBodyFolderResumeErrorAndValidEmptyBodies(t *testing.T) { + var refuse atomic.Bool + c, _ := interceptingServer(t, nil, func(line string) string { + if refuse.Load() && strings.Contains(strings.ToUpper(line), "EXAMINE") { + return strings.Fields(line)[0] + " NO folder unavailable" + } + return "" + }, "Archive") + if err := c.Connect(); err != nil { + t.Fatal(err) + } + uid := putMessage(t, c) + if _, err := c.SelectForSync("INBOX"); err != nil { + t.Fatal(err) + } + fetched, err := c.FetchEnvelopes(t.Context(), []goimap.UID{uid}) + if err != nil || len(fetched) != 1 { + t.Fatal("envelope failed") + } + if _, err := c.selectMailbox("Archive", nil); err != nil { + t.Fatal(err) + } + refuse.Store(true) + if err := c.FetchBody(fetched[0]); err == nil { + t.Fatal("failed folder resume swallowed") + } + refuse.Store(false) + if err := c.FetchBody(fetched[0]); err != nil { + t.Fatal(err) + } + for _, bs := range []goimap.BodyStructure{nil, &goimap.BodyStructureSinglePart{Type: "application", Subtype: "octet-stream"}, &goimap.BodyStructureSinglePart{Type: "text", Subtype: "plain", Extended: &goimap.BodyStructureSinglePartExt{Disposition: &goimap.BodyStructureDisposition{Value: "attachment"}}}} { + f := &Fetched{Email: &models.EmailMessageData{}, uid: uid, body: bs} + if err := c.FetchBody(f); err != nil || f.Email.BodyPlain != "" || f.Email.BodyHTML != "" { + t.Fatalf("valid no-text message refused: %v", err) + } + } +} + +func TestNomodseqWireOmitsModseqAndChangedSince(t *testing.T) { + c, wire := bodyRepairFixture(t, nil) + if err := c.Connect(); err != nil { + t.Fatal(err) + } + if !c.HasCondStore() { + t.Fatal("fixture did not advertise CONDSTORE") + } + view, err := c.SelectForSyncState("INBOX") + if err != nil || view.HighestModSeq != 0 { + t.Fatalf("NOMODSEQ SELECT: %+v %v", view, err) + } + if _, err := c.FetchEnvelopes(t.Context(), []goimap.UID{1}); err != nil { + t.Fatal(err) + } + if uids, err := c.SearchNewSince(1); err != nil || len(uids) != 1 { + t.Fatalf("UIDNEXT fallback: %v %v", uids, err) + } + flags, err := c.FetchFlags(t.Context(), 1) + if err != nil || len(flags) != 1 { + t.Fatalf("flag fallback: %v %v", flags, err) + } + for _, line := range wire.commands("FETCH") { + if strings.Contains(line, "MODSEQ") || strings.Contains(line, "CHANGEDSINCE") { + t.Fatalf("NOMODSEQ command: %s", line) + } + } + for _, line := range wire.commands("SEARCH") { + if strings.Contains(line, "MODSEQ") || strings.Contains(line, "CHANGEDSINCE") { + t.Fatalf("NOMODSEQ search: %s", line) + } + } +} + +func TestSetupBlackholesAreBoundedAndCanRedial(t *testing.T) { + for _, phase := range []string{"greeting", "STARTTLS response", "STARTTLS handshake", "implicit TLS handshake", "authentication"} { + t.Run(phase, func(t *testing.T) { + var attempts atomic.Int32 + wire := &wireLog{} + addr := repairWireServer(t, func(conn net.Conn) { + attempt := attempts.Add(1) + if attempt == 1 && (phase == "greeting" || phase == "implicit TLS handshake") { + _, _ = io.Copy(io.Discard, conn) + return + } + _, _ = io.WriteString(conn, "* OK [CAPABILITY IMAP4rev1 STARTTLS] ready\r\n") + r := bufio.NewReader(conn) + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + wire.add(strings.TrimSpace(line)) + fields := strings.Fields(line) + if len(fields) < 2 { + return + } + tag := fields[0] + if attempt == 1 { + if phase == "STARTTLS handshake" { + _, _ = fmt.Fprintf(conn, "%s OK begin TLS\r\n", tag) + } + _, _ = io.Copy(io.Discard, conn) + return + } + switch strings.ToUpper(fields[1]) { + case "LOGIN": + _, _ = fmt.Fprintf(conn, "%s OK [CAPABILITY IMAP4rev1] logged in\r\n", tag) + case "CAPABILITY": + _, _ = fmt.Fprintf(conn, "* CAPABILITY IMAP4rev1\r\n%s OK capability\r\n", tag) + default: + _, _ = fmt.Fprintf(conn, "%s BAD unexpected command\r\n", tag) + } + } + }) + c := clientTo(t, addr) + c.plaintext = phase == "authentication" + c.Credentials.Security = models.MailSecurityStartTLS + if phase == "implicit TLS handshake" { + c.Credentials.Security = models.MailSecurityTLS + } + c.IdleTimeout = 100 * time.Millisecond + start := time.Now() + if err := c.Connect(); err == nil { + t.Fatal("blackhole succeeded") + } + if time.Since(start) > 2*time.Second { + t.Fatal("setup budget did not bound silent peer") + } + if phase != "authentication" && len(wire.commands("LOGIN")) != 0 { + t.Fatal("TLS failure fell back to plaintext authentication") + } + // The second local connection uses test-only plaintext to isolate redial from certificate trust. + c.plaintext = true + if err := c.ensureConnected(); err != nil { + t.Fatalf("redial: %v", err) + } + if attempts.Load() != 2 { + t.Fatalf("connections = %d", attempts.Load()) + } + }) + } +} + +func TestCloseAndLifetimeCancellationInterruptPendingSetup(t *testing.T) { + for _, explicit := range []bool{false, true} { + t.Run(fmt.Sprintf("explicit close=%t", explicit), func(t *testing.T) { + accepted := make(chan struct{}) + ended := make(chan struct{}) + addr := repairWireServer(t, func(conn net.Conn) { close(accepted); _, _ = io.Copy(io.Discard, conn); close(ended) }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + c := clientTo(t, addr) + c.plaintext = false + c.Context = ctx + c.IdleTimeout = 10 * time.Second + c.Credentials.Security = models.MailSecurityStartTLS + result := make(chan *errx.MailError, 1) + go func() { result <- c.Connect() }() + select { + case <-accepted: + case <-time.After(time.Second): + t.Fatal("setup did not connect") + } + if explicit { + if err := c.Close(); err != nil { + t.Fatal(err) + } + } else { + cancel() + } + select { + case err := <-result: + if err == nil { + t.Fatal("cancelled setup succeeded") + } + case <-time.After(time.Second): + t.Fatal("pending setup did not cancel") + } + select { + case <-ended: + case <-time.After(time.Second): + t.Fatal("pending socket did not close") + } + if err := c.ensureConnected(); err == nil { + t.Fatal("retired lifetime redialed") + } + }) + } +} + +func TestCloseAndLifetimeCancellationCloseIdleSessionWithoutRedial(t *testing.T) { + for _, explicit := range []bool{false, true} { + t.Run(fmt.Sprintf("explicit close=%t", explicit), func(t *testing.T) { + ended := make(chan struct{}) + var attempts atomic.Int32 + addr := repairWireServer(t, func(conn net.Conn) { + attempts.Add(1) + _, _ = io.WriteString(conn, "* OK [CAPABILITY IMAP4rev1] ready\r\n") + r := bufio.NewReader(conn) + for { + line, err := r.ReadString('\n') + if err != nil { + close(ended) + return + } + _, _ = fmt.Fprintf(conn, "%s OK [CAPABILITY IMAP4rev1] authenticated\r\n", strings.Fields(line)[0]) + } + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + c := clientTo(t, addr) + c.Context = ctx + if err := c.Connect(); err != nil { + t.Fatal(err) + } + if explicit { + if err := c.Close(); err != nil { + t.Fatal(err) + } + if err := c.Close(); err != nil { + t.Fatal(err) + } + } else { + cancel() + } + select { + case <-ended: + case <-time.After(time.Second): + t.Fatal("idle server did not observe socket EOF") + } + if err := c.ensureConnected(); err == nil || attempts.Load() != 1 { + t.Fatal("retired installed session reconnected") + } + }) + } +} + +func TestStartTLSSetupStillRejectsUntrustedCertificate(t *testing.T) { + t.Setenv("MAIL_TLS_INSECURE", "false") + certServer := httptest.NewTLSServer(nil) + defer certServer.Close() + wire := &wireLog{} + addr := repairWireServer(t, func(conn net.Conn) { + _, _ = io.WriteString(conn, "* OK [CAPABILITY IMAP4rev1 STARTTLS] ready\r\n") + line, err := bufio.NewReader(conn).ReadString('\n') + if err != nil { + return + } + wire.add(strings.TrimSpace(line)) + _, _ = fmt.Fprintf(conn, "%s OK begin TLS\r\n", strings.Fields(line)[0]) + secure := tls.Server(conn, certServer.TLS) + if err := secure.Handshake(); err == nil { + line, _ := bufio.NewReader(secure).ReadString('\n') + wire.add(strings.TrimSpace(line)) + } + }) + c := clientTo(t, addr) + c.plaintext = false + c.Credentials.Security = models.MailSecurityStartTLS + c.IdleTimeout = time.Second + if err := c.Connect(); err == nil { + t.Fatal("untrusted fixture certificate was accepted") + } + if len(wire.commands("LOGIN")) != 0 { + t.Fatal("certificate failure sent plaintext credentials") + } +} + +func TestFetchBodyTLSLiteralFailureReleasesDecoderAndRetries(t *testing.T) { + var attempts atomic.Int32 + c, _ := bodyRepairFixture(t, func(conn net.Conn, tag string) bool { + if attempts.Add(1) == 1 { + _, _ = io.WriteString(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\nshort") + return false + } + _, _ = fmt.Fprintf(conn, "* 1 FETCH (UID 1 BODY[1] {13}\r\ncomplete body)\r\n%s OK body\r\n", tag) + return true + }) + certServer := httptest.NewTLSServer(nil) + defer certServer.Close() + backendAddr := net.JoinHostPort(c.Credentials.Host, fmt.Sprint(c.Credentials.Port)) + addr := repairWireServer(t, func(raw net.Conn) { + secure := tls.Server(raw, certServer.TLS) + upstream, err := net.DialTimeout("tcp", backendAddr, time.Second) + if err != nil { + return + } + defer func() { _ = upstream.Close() }() + go func() { _, _ = io.Copy(upstream, secure); _ = upstream.Close() }() + _, _ = io.Copy(secure, upstream) + // Close the raw transport without TLS close_notify to produce TLS unexpected EOF. + }) + roots := x509.NewCertPool() + roots.AddCert(certServer.Certificate()) + connect := func() { + raw, err := net.DialTimeout("tcp", addr, time.Second) + if err != nil { + t.Fatal(err) + } + conn := &idleConn{Conn: raw, timeout: time.Second} + secure := tls.Client(conn, &tls.Config{RootCAs: roots, ServerName: certServer.Certificate().DNSNames[0], MinVersion: tls.VersionTLS12}) + if err := secure.HandshakeContext(t.Context()); err != nil { + _ = raw.Close() + t.Fatal(err) + } + c.client = imapclient.New(secure, nil) + c.conn = conn + c.transport = raw + c.selected.Store(false) + c.selection.Store(nil) + if err := c.plainAuth(); err != nil { + t.Fatal(err) + } + c.condStore.Store(c.client.Caps().Has(goimap.CapCondStore)) + if _, err := c.SelectForSync("INBOX"); err != nil { + t.Fatal(err) + } + } + connect() + fetched, err := c.FetchEnvelopes(t.Context(), []goimap.UID{1}) + if err != nil || len(fetched) != 1 { + t.Fatalf("TLS envelope: %v %v", fetched, err) + } + result := make(chan *errx.MailError, 1) + go func() { result <- c.FetchBody(fetched[0]) }() + select { + case err := <-result: + if err == nil { + t.Fatal("TLS literal failure swallowed") + } + case <-time.After(time.Second): + t.Fatal("TLS literal error left decoder blocked") + } + connect() + fetched, err = c.FetchEnvelopes(t.Context(), []goimap.UID{1}) + if err != nil || len(fetched) != 1 { + t.Fatalf("TLS retry envelope: %v %v", fetched, err) + } + if err := c.FetchBody(fetched[0]); err != nil || fetched[0].Email.BodyPlain != "complete body" { + t.Fatalf("TLS retry body: %q %v", fetched[0].Email.BodyPlain, err) + } +}