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

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

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=

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

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
 			}

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

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: "<known@test>"}}}
+			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: "<known@test>", 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: "<one@test>"}, 2: {MessageID: "<two@test>"}}}
+	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")
+	}
+}

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

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

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

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

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

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 \"<wire@test>\") 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 \"<wire@test>\"))\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)
+	}
+}