From e6210dff09c40e29aeb3fc658ab76506d9f037e4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 28 Aug 2026 00:40:18 -0700 Subject: [PATCH] feat: stop a transient Graph failure on a folder's first backfill page from marking that folder's backfill permanently complete: msgraph.HandleError mapped 404, every 5xx and any unrecognised status onto the same ErrMailServerUnreachable, so graphBackfill's "the tenant does not have this folder" skip fired on a 503 as well and wrote the folder off in a cursor that is persisted through SYNC_STATE and handed back on every later load, silently costing a customer who connected a mailbox during a Graph incident their archive history; 404 now carries its own RESOURCE_NOT_FOUND code, the skip keys on that alone, a genuine unreachable server ends the pass with the folder's cursor held so the next one retries it, and the Gmail and IMAP imports get the same regression shape pinned by tests --- internal/app/worker/health_record.go | 3 +- internal/app/worker/wmail/err.go | 1 + internal/app/worker/wmail/send.go | 2 +- internal/app/worker/wmail/sync_google_test.go | 107 ++++++ internal/app/worker/wmail/sync_graph.go | 12 +- internal/app/worker/wmail/sync_graph_test.go | 344 ++++++++++++++++++ internal/app/worker/wmail/sync_imap_test.go | 107 ++++++ internal/client/msgraph/error.go | 26 +- internal/client/msgraph/error_test.go | 103 ++++++ internal/errx/email.go | 10 +- 10 files changed, 701 insertions(+), 14 deletions(-) create mode 100644 internal/app/worker/wmail/sync_graph_test.go create mode 100644 internal/client/msgraph/error_test.go diff --git a/internal/app/worker/health_record.go b/internal/app/worker/health_record.go index 54d6c862..5dc9f5ba 100644 --- a/internal/app/worker/health_record.go +++ b/internal/app/worker/health_record.go @@ -47,7 +47,8 @@ func (s *WorkerService) recordSendOutcome(result *wmail.SendResult) { errx.MailErrorCodeAccountSuspended: s.RecordBounceHard() case errx.MailErrorCodeServerUnreachable, - errx.MailErrorCodeConnectionLost: + errx.MailErrorCodeConnectionLost, + errx.MailErrorCodeNotFound: s.RecordBounceSoft() default: // Best-effort classification on free-text — keeps the signal diff --git a/internal/app/worker/wmail/err.go b/internal/app/worker/wmail/err.go index 52e3a06c..f46c31f4 100644 --- a/internal/app/worker/wmail/err.go +++ b/internal/app/worker/wmail/err.go @@ -74,6 +74,7 @@ func mailErrorToJobEventType(mailErr *errx.MailError) models.JobEventType { // event would deactivate the mailbox for a transient provider throttle. case errx.MailErrorCodeServerUnreachable, errx.MailErrorCodeConnectionLost, + errx.MailErrorCodeNotFound, errx.MailErrorCodeImapUnknown: return models.JobEventTypeEmailServerError } diff --git a/internal/app/worker/wmail/send.go b/internal/app/worker/wmail/send.go index 2c77f153..ad8d56ff 100644 --- a/internal/app/worker/wmail/send.go +++ b/internal/app/worker/wmail/send.go @@ -440,7 +440,7 @@ func DetermineErrorEventType(err *errx.MailError) models.JobEventType { case errx.MailErrorCodeRateLimitExceeded, errx.MailErrorCodeSendingTooFast, errx.MailErrorCodeQuotaExceeded: return models.JobEventTypeEmailRateLimited - case errx.MailErrorCodeServerUnreachable, errx.MailErrorCodeConnectionLost: + case errx.MailErrorCodeServerUnreachable, errx.MailErrorCodeConnectionLost, errx.MailErrorCodeNotFound: return models.JobEventTypeEmailServerError default: diff --git a/internal/app/worker/wmail/sync_google_test.go b/internal/app/worker/wmail/sync_google_test.go index 0ff64ce2..8aba918b 100644 --- a/internal/app/worker/wmail/sync_google_test.go +++ b/internal/app/worker/wmail/sync_google_test.go @@ -1,10 +1,18 @@ package wmail import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" "testing" + "time" "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/client/goog" "github.com/warmbly/warmbly/internal/models" + "golang.org/x/oauth2" ) // The checkpoint row is keyed (user_id, email_id) with a foreign key to users, @@ -47,3 +55,102 @@ func TestNewHistoryIDCarriesRowKey(t *testing.T) { t.Error("event carries a nil UUID, which violates the users foreign key") } } + +// fakeGmail is the Gmail API as one backfill sees it: a messages.list that can +// be made to refuse, and a messages.get for whatever it did list. +type fakeGmail struct { + ids []string + failList int // list calls left to refuse (-1 for always) + listCalls int +} + +func (g *fakeGmail) serve(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + id, isGet := strings.CutPrefix(r.URL.Path, "/gmail/v1/users/me/messages/") + if isGet { + _, _ = fmt.Fprintf(w, `{"id":%q,"threadId":"t-%s","payload":{"headers":[{"name":"Message-Id","value":"<%s@gmail.test>"},{"name":"Subject","value":"history"}]}}`, id, id, id) + return + } + if g.failList != 0 { + if g.failList > 0 { + g.failList-- + } + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"code":400,"message":"injected"}}`)) + return + } + g.listCalls++ + msgs := make([]string, 0, len(g.ids)) + for _, id := range g.ids { + msgs = append(msgs, fmt.Sprintf(`{"id":%q,"threadId":"t-%s"}`, id, id)) + } + _, _ = fmt.Fprintf(w, `{"messages":[%s]}`, strings.Join(msgs, ",")) + })) + t.Cleanup(srv.Close) + return srv +} + +func newGoogleTestMail(t *testing.T, srv *httptest.Server, events *[]captured) *WMail { + t.Helper() + w := &WMail{ + ID: uuid.New(), + UserID: uuid.New(), + Email: "box@gmail.test", + EmailType: models.InboxProviderGoogle, + Storage: fakeStore{}, + EmailMessageMapRepository: fakeMessageMap{}, + gov: newGovernor(uuid.New(), nil, nil, models.SyncPolicy{}), + } + w.onEvent = func(kind models.JobEventType, body any) error { + *events = append(*events, captured{eventType: kind, body: body}) + return nil + } + w.tracker = newSyncTracker(nil, func(models.SyncState) error { return nil }) + + client := &goog.Client{ + Email: w.Email, + OnMessageAdded: w.onGoogleMessageAdded, + OnTokenRefresh: func(context.Context, *oauth2.Token) error { return nil }, + } + ctx := context.WithValue(context.Background(), oauth2.HTTPClient, &http.Client{Transport: rewriteToTestServer(srv.URL)}) + token := &oauth2.Token{AccessToken: "live", Expiry: time.Now().Add(time.Hour)} + if merr := client.Init(ctx, token, oauth2.Config{}); merr != nil { + t.Fatalf("client init: %v", merr.Message) + } + w.GoogleData = &GoogleData{Client: client} + return w +} + +// The Graph defect's shape, checked on the Gmail import: a refused listing +// ends the pass with the page token where it was, and never reports the +// history as imported. +func TestGoogleBackfillRetriesAfterATransientFailure(t *testing.T) { + g := &fakeGmail{ids: []string{"g1", "g2"}, failList: 1} + var events []captured + w := newGoogleTestMail(t, g.serve(t), &events) + + if merr := w.googleBackfill(t.Context(), &tickStats{}); merr == nil { + t.Fatal("a refused listing was swallowed; the pass must end so the import is retried") + } + if st := w.tracker.state.BackfillStatus; st == models.SyncBackfillComplete { + t.Fatalf("backfill status = %s after a failed listing", st) + } + if tok := w.tracker.state.BackfillCursor.PageToken; tok != "" { + t.Errorf("page token = %q, want it held where it was", tok) + } + + if merr := w.googleBackfill(t.Context(), &tickStats{}); merr != nil { + t.Fatalf("second pass: %v", merr.Message) + } + if st := w.tracker.state.BackfillStatus; st != models.SyncBackfillComplete { + t.Fatalf("backfill status = %s, want %s", st, models.SyncBackfillComplete) + } + if got := len(importedIDs(events)); got != 2 { + t.Errorf("imported %d messages, want 2: %v", got, importedIDs(events)) + } + if g.listCalls != 1 { + t.Errorf("listed %d times, want the one call that succeeded", g.listCalls) + } +} diff --git a/internal/app/worker/wmail/sync_graph.go b/internal/app/worker/wmail/sync_graph.go index d46b4c05..1d0d7bcd 100644 --- a/internal/app/worker/wmail/sync_graph.go +++ b/internal/app/worker/wmail/sync_graph.go @@ -6,6 +6,7 @@ import ( "time" "github.com/google/uuid" + "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/client/msgraph" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" @@ -167,9 +168,14 @@ func (w *WMail) graphBackfill(ctx context.Context, stats *tickStats) *errx.MailE if err != nil { var mailErr *errx.MailError if errors.As(err, &mailErr) { - // A folder the tenant does not have (archive on some - // plans) is skipped, not fatal. - if mailErr.Code == errx.MailErrorCodeServerUnreachable && cur.Next == "" { + // Only Graph saying the folder is absent (archive on + // some plans) skips it; a 503 ends the pass instead, or + // one blip marks the folder complete forever. + if mailErr.Code == errx.MailErrorCodeNotFound { + log.Debug(). + Str("email_id", w.ID.String()). + Str("folder", folder). + Msg("backfill: folder absent on the tenant, skipped") w.tracker.setFolder(folder, models.SyncFolderCursor{Done: true}) break } diff --git a/internal/app/worker/wmail/sync_graph_test.go b/internal/app/worker/wmail/sync_graph_test.go new file mode 100644 index 00000000..c4f7e650 --- /dev/null +++ b/internal/app/worker/wmail/sync_graph_test.go @@ -0,0 +1,344 @@ +package wmail + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/client/msgraph" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "golang.org/x/oauth2" +) + +// fakeGraph is Microsoft Graph as one mailbox's sync pass sees it: a delta +// stream per tracked folder, a listing per backfill folder, and per-folder +// failure injection so a pass can be driven through a provider incident. +type fakeGraph struct { + // messages is what each folder's listing returns, by message id. + messages map[string][]string + // live is offered once through the inbox delta stream. + live []string + // listFail is the status a folder's listing answers with, and how many + // calls answer that way before it recovers. + listFail map[string]*graphFailure + // listed counts successful listings per folder. + listed map[string]int +} + +type graphFailure struct { + status int + times int +} + +func newFakeGraph() *fakeGraph { + return &fakeGraph{ + messages: map[string][]string{}, + listFail: map[string]*graphFailure{}, + listed: map[string]int{}, + } +} + +func (g *fakeGraph) serve(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + path := strings.TrimPrefix(r.URL.Path, "/v1.0/me/") + + // Hydration of one live message: GET /me/messages/{id} + if id, ok := strings.CutPrefix(path, "messages/"); ok { + _ = json.NewEncoder(w).Encode(graphMessageJSON(id)) + return + } + + rest, ok := strings.CutPrefix(path, "mailFolders/") + if !ok { + t.Errorf("unexpected Graph path %s", r.URL.Path) + w.WriteHeader(http.StatusNotImplemented) + return + } + folder, tail, _ := strings.Cut(rest, "/") + + if tail == "messages/delta" { + value := []any{} + if folder == msgraph.FolderInbox { + for _, id := range g.live { + value = append(value, map[string]any{"id": id, "isRead": false}) + } + g.live = nil + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "value": value, + "@odata.deltaLink": "https://graph.microsoft.com/v1.0/me/mailFolders/" + folder + "/messages/delta?$deltatoken=t", + "@odata.deltaToken": "t", + }) + return + } + + if f := g.listFail[folder]; f != nil && f.times != 0 { + if f.times > 0 { + f.times-- + } + w.WriteHeader(f.status) + _, _ = w.Write([]byte(`{"error":{"code":"Failed","message":"injected"}}`)) + return + } + + g.listed[folder]++ + value := make([]any, 0, len(g.messages[folder])) + for _, id := range g.messages[folder] { + value = append(value, graphMessageJSON(id)) + } + _ = json.NewEncoder(w).Encode(map[string]any{"value": value}) + })) + t.Cleanup(srv.Close) + return srv +} + +func graphMessageJSON(id string) map[string]any { + return map[string]any{ + "id": id, + "internetMessageId": fmt.Sprintf("<%s@outlook.test>", id), + "conversationId": "conv-" + id, + "subject": "subject " + id, + "receivedDateTime": time.Now().UTC().Format(time.RFC3339), + "from": map[string]any{"emailAddress": map[string]any{"address": "someone@example.test"}}, + "body": map[string]any{"contentType": "text", "content": "body " + id}, + } +} + +// rewriteToTestServer sends the real provider URLs a client builds at the test +// server instead, keeping the path and query intact. +func rewriteToTestServer(base string) http.RoundTripper { + target, err := url.Parse(base) + if err != nil { + panic(err) + } + return testRoundTripFunc(func(req *http.Request) (*http.Response, error) { + clone := req.Clone(req.Context()) + clone.URL.Scheme = target.Scheme + clone.URL.Host = target.Host + clone.Host = target.Host + return http.DefaultTransport.RoundTrip(clone) + }) +} + +type testRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f testRoundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +// newGraphTestMail wires a real msgraph client, pointed at the fake tenant, +// into the smallest WMail that can run a Graph sync pass. +func newGraphTestMail(t *testing.T, srv *httptest.Server, events *[]captured, relayed *[]models.SyncState) *WMail { + t.Helper() + w := &WMail{ + ID: uuid.New(), + UserID: uuid.New(), + Email: "box@outlook.test", + EmailType: models.InboxProviderOutlook, + Storage: fakeStore{}, + EmailMessageMapRepository: fakeMessageMap{}, + gov: newGovernor(uuid.New(), nil, nil, models.SyncPolicy{}), + } + w.onEvent = func(kind models.JobEventType, body any) error { + *events = append(*events, captured{eventType: kind, body: body}) + return nil + } + w.tracker = newSyncTracker(nil, func(st models.SyncState) error { + *relayed = append(*relayed, st) + return nil + }) + + // Seeded cursors: a folder with no delta link is primed, not imported, so + // without these the live half of the pass would never run. + deltaLinks := map[string]string{} + for _, folder := range msgraph.TrackedFolders { + deltaLinks[folder] = "https://graph.microsoft.com/v1.0/me/mailFolders/" + folder + "/messages/delta?$deltatoken=seed" + } + client := &msgraph.Client{ + Email: w.Email, + DeltaLinks: deltaLinks, + OnMessageSeen: w.onGraphMessageSeen, + OnMessageRemove: w.onGraphMessageRemove, + OnDelta: w.onGraphDelta, + OnTokenRefresh: func(context.Context, *oauth2.Token) error { return nil }, + } + ctx := context.WithValue(context.Background(), oauth2.HTTPClient, &http.Client{Transport: rewriteToTestServer(srv.URL)}) + token := &oauth2.Token{AccessToken: "live", Expiry: time.Now().Add(time.Hour)} + if merr := client.Init(ctx, token, oauth2.Config{}); merr != nil { + t.Fatalf("client init: %v", merr.Message) + } + w.GraphData = &GraphData{Client: client} + return w +} + +// importedIDs is every message the pass stored, in order. +func importedIDs(events []captured) []string { + var out []string + for _, e := range events { + if e.eventType != models.JobEventTypeNewEmail { + continue + } + if ev, ok := e.body.(*models.JobEventNewEmail); ok { + out = append(out, ev.Message.GmailID) + } + } + return out +} + +func hasID(ids []string, want string) bool { + for _, id := range ids { + if id == want { + return true + } + } + return false +} + +// The incident: one Graph blip on a folder's first backfill page used to mark +// that folder complete forever, so the customer who connected a mailbox during +// an outage silently got no archive history, on this worker or any later one. +func TestGraphBackfillRetriesAFolderAfterATransientFailure(t *testing.T) { + g := newFakeGraph() + g.messages[msgraph.FolderInbox] = []string{"inbox-1"} + g.messages[msgraph.FolderSent] = []string{"sent-1"} + g.messages[msgraph.FolderArchive] = []string{"archive-1"} + g.live = []string{"live-1"} + // Graph is having a moment, but only for the archive listing. + g.listFail[msgraph.FolderArchive] = &graphFailure{status: http.StatusServiceUnavailable, times: 1} + + var events []captured + var relayed []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &relayed) + + merr := w.SyncGraph(t.Context()) + if merr == nil { + t.Fatal("a 503 on the archive listing was swallowed; the pass must end so the folder is retried") + } + if merr.Code != errx.MailErrorCodeServerUnreachable { + t.Fatalf("code = %s, want %s", merr.Code, errx.MailErrorCodeServerUnreachable) + } + if cur := w.tracker.folder(msgraph.FolderArchive); cur.Done { + t.Fatal("the archive backfill was marked complete by a transient failure") + } + if st := w.tracker.state.BackfillStatus; st == models.SyncBackfillComplete { + t.Fatalf("backfill status = %s after a failed pass", st) + } + // The persisted half of the same defect: nothing that reaches the control + // plane may write the folder off, or a replaced worker never retries it. + for _, st := range relayed { + if st.BackfillCursor.Folders[msgraph.FolderArchive].Done { + t.Fatal("a relayed SYNC_STATE marked the archive folder done after a transient failure") + } + } + // The folders that did answer still landed, and live mail was not lost. + for _, want := range []string{"live-1", "inbox-1", "sent-1"} { + if !hasID(importedIDs(events), want) { + t.Errorf("%s was not imported: %v", want, importedIDs(events)) + } + } + + // Next pass: Graph is back. + if merr := w.SyncGraph(t.Context()); merr != nil { + t.Fatalf("second pass: %v", merr.Message) + } + if !hasID(importedIDs(events), "archive-1") { + t.Fatalf("the archive history was never imported: %v", importedIDs(events)) + } + if !w.tracker.folder(msgraph.FolderArchive).Done { + t.Error("archive is still not done after a successful listing") + } + + // One more pass settles the whole backfill and relays it. + if merr := w.SyncGraph(t.Context()); merr != nil { + t.Fatalf("third pass: %v", merr.Message) + } + if st := w.tracker.state.BackfillStatus; st != models.SyncBackfillComplete { + t.Fatalf("backfill status = %s, want %s", st, models.SyncBackfillComplete) + } + if len(relayed) == 0 { + t.Fatal("no SYNC_STATE was relayed, so nothing would be persisted") + } + last := relayed[len(relayed)-1] + if !last.BackfillCursor.Folders[msgraph.FolderArchive].Done { + t.Error("the relayed state does not carry the archive folder as done") + } +} + +// The behavior the skip was written for, now keyed on Graph actually saying so: +// a tenant without an archive folder is not a mailbox that syncs forever. +func TestGraphBackfillSkipsAFolderTheTenantDoesNotHave(t *testing.T) { + g := newFakeGraph() + g.messages[msgraph.FolderInbox] = []string{"inbox-1"} + g.messages[msgraph.FolderSent] = []string{"sent-1"} + // No archive on this plan: Graph answers 404 for as long as it is asked. + g.listFail[msgraph.FolderArchive] = &graphFailure{status: http.StatusNotFound, times: -1} + + var events []captured + var relayed []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &relayed) + + if merr := w.SyncGraph(t.Context()); merr != nil { + t.Fatalf("a missing folder must not fail the pass: %v", merr.Message) + } + if !w.tracker.folder(msgraph.FolderArchive).Done { + t.Fatal("the absent archive folder was not skipped, so the backfill can never finish") + } + for _, want := range []string{"inbox-1", "sent-1"} { + if !hasID(importedIDs(events), want) { + t.Errorf("%s was not imported: %v", want, importedIDs(events)) + } + } + + if merr := w.SyncGraph(t.Context()); merr != nil { + t.Fatalf("second pass: %v", merr.Message) + } + if st := w.tracker.state.BackfillStatus; st != models.SyncBackfillComplete { + t.Fatalf("backfill status = %s, want %s", st, models.SyncBackfillComplete) + } + if g.listed[msgraph.FolderArchive] != 0 { + t.Error("the fake tenant served an archive listing it was supposed to refuse") + } +} + +// Every folder is retried on its own terms: a blip on the inbox listing holds +// the inbox, and does not quietly hand the sent folder the same verdict. +func TestGraphBackfillHoldsOnlyTheFolderThatFailed(t *testing.T) { + g := newFakeGraph() + g.messages[msgraph.FolderInbox] = []string{"inbox-1"} + g.messages[msgraph.FolderSent] = []string{"sent-1"} + g.messages[msgraph.FolderArchive] = []string{"archive-1"} + g.listFail[msgraph.FolderInbox] = &graphFailure{status: http.StatusInternalServerError, times: 1} + + var events []captured + var relayed []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &relayed) + + if merr := w.SyncGraph(t.Context()); merr == nil { + t.Fatal("a 500 on the inbox listing was swallowed") + } + if w.tracker.folder(msgraph.FolderInbox).Done { + t.Fatal("the inbox backfill was marked complete by a transient failure") + } + // The pass ended at the inbox, so the folders behind it are untouched, not + // written off. + if w.tracker.folder(msgraph.FolderSent).Done || w.tracker.folder(msgraph.FolderArchive).Done { + t.Fatal("a later folder was marked done by a failure in an earlier one") + } + + if merr := w.SyncGraph(t.Context()); merr != nil { + t.Fatalf("second pass: %v", merr.Message) + } + for _, want := range []string{"inbox-1", "sent-1", "archive-1"} { + if !hasID(importedIDs(events), want) { + t.Errorf("%s was not imported: %v", want, importedIDs(events)) + } + } +} diff --git a/internal/app/worker/wmail/sync_imap_test.go b/internal/app/worker/wmail/sync_imap_test.go index 11ce0862..ebb98ecb 100644 --- a/internal/app/worker/wmail/sync_imap_test.go +++ b/internal/app/worker/wmail/sync_imap_test.go @@ -236,3 +236,110 @@ func TestImapSyncWalksEveryBatchWithinBudget(t *testing.T) { t.Errorf("released the mailbox %d times, want once before LIST-STATUS", conn.released) } } + +// backfillImapConn serves an initial import: per-folder UID lists, with a +// folder's search made to fail on demand so a pass can be driven through a +// server having a moment. +type backfillImapConn struct { + ImapConn + folders []models.Mailbox + uids map[string][]goimap.UID + fail map[string]int // folder -> failures left (-1 for always) + selected string +} + +func (c *backfillImapConn) Folders() ([]models.Mailbox, *errx.MailError) { return c.folders, nil } +func (c *backfillImapConn) ReleaseMailbox() {} + +func (c *backfillImapConn) SelectForSync(name string) (uint32, *errx.MailError) { + c.selected = name + return uint32(len(c.uids[name])), nil +} + +func (c *backfillImapConn) SearchChangedSince(uint64) ([]goimap.UID, *errx.MailError) { + return nil, nil +} + +func (c *backfillImapConn) SearchSince(time.Time) ([]goimap.UID, *errx.MailError) { + if n := c.fail[c.selected]; n != 0 { + if n > 0 { + c.fail[c.selected] = n - 1 + } + return nil, errx.ErrMailServerUnreachable + } + return append([]goimap.UID(nil), c.uids[c.selected]...), nil +} + +func (c *backfillImapConn) FetchEnvelopes(_ context.Context, uids []goimap.UID) ([]*imap.Fetched, *errx.MailError) { + out := make([]*imap.Fetched, 0, len(uids)) + for _, uid := range uids { + out = append(out, &imap.Fetched{Email: &models.EmailMessageData{ + UID: uint32(uid), + MessageID: fmt.Sprintf("<%s-%d@fake.test>", c.selected, uid), + Subject: "history", + }}) + } + return out, nil +} + +func (c *backfillImapConn) FetchBody(*imap.Fetched) {} + +// 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 +// server error may mark a folder, or the whole import, complete. +func TestImapBackfillRetriesAFolderAfterATransientFailure(t *testing.T) { + conn := &backfillImapConn{ + folders: []models.Mailbox{ + {Name: "INBOX", UIDValidity: 7, HighestModSeq: 100}, + {Name: "Archive", UIDValidity: 8, HighestModSeq: 100}, + }, + uids: map[string][]goimap.UID{"INBOX": {5, 6}, "Archive": {9}}, + fail: map[string]int{"Archive": 1}, + } + var events []captured + w := &WMail{ + UserID: uuid.New(), + ID: uuid.New(), + Storage: fakeStore{}, + EmailMessageMapRepository: fakeMessageMap{}, + gov: newGovernor(uuid.New(), nil, nil, models.SyncPolicy{}), + SmtpImapData: &SmtpImapData{ + ImapClient: conn, + Mailboxes: []*models.Mailbox{ + {Name: "INBOX", UIDValidity: 7, HighestModSeq: 100}, + {Name: "Archive", UIDValidity: 8, HighestModSeq: 100}, + }, + }, + } + w.onEvent = func(kind models.JobEventType, body any) error { + events = append(events, captured{eventType: kind, body: body}) + return nil + } + w.tracker = newSyncTracker(nil, func(models.SyncState) error { return nil }) + + if err := w.Sync(t.Context()); err == nil { + t.Fatal("a failed folder search was swallowed; the pass must end so the folder is retried") + } + if w.tracker.folder("8").Done { + t.Fatal("the archive backfill was marked complete by a transient failure") + } + if st := w.tracker.state.BackfillStatus; st == models.SyncBackfillComplete { + t.Fatalf("backfill status = %s after a failed pass", st) + } + + if err := w.Sync(t.Context()); err != nil { + t.Fatalf("second pass: %v", err.Message) + } + if !w.tracker.folder("8").Done { + t.Error("archive is still not done after a successful search") + } + if err := w.Sync(t.Context()); err != nil { + t.Fatalf("third pass: %v", err.Message) + } + if st := w.tracker.state.BackfillStatus; st != models.SyncBackfillComplete { + t.Fatalf("backfill status = %s, want %s", st, models.SyncBackfillComplete) + } + if !hasEvent(events, models.JobEventTypeNewEmail) { + t.Error("no history was imported at all") + } +} diff --git a/internal/client/msgraph/error.go b/internal/client/msgraph/error.go index 57bf9544..680e0aad 100644 --- a/internal/client/msgraph/error.go +++ b/internal/client/msgraph/error.go @@ -22,6 +22,9 @@ type graphErrorEnvelope struct { // critical (needs re-auth / stop) failures: // - 401 -> authentication failed (token expired/revoked, re-consent) // - 403 -> authorization failed (missing scope / mailbox disabled) +// - 404 -> resource not found (a folder the tenant does not have, a message +// already gone); its own code so a caller can skip what is absent without +// also skipping on a 503 // - 429 -> sending too fast (throttled; Retry-After honored by the caller loop) // - 5xx / other -> server unreachable (retry) func HandleError(resp *http.Response) *errx.MailError { @@ -29,19 +32,26 @@ func HandleError(resp *http.Response) *errx.MailError { var env graphErrorEnvelope _ = json.Unmarshal(body, &env) - switch resp.StatusCode { - case http.StatusUnauthorized: - return errx.ErrMailAuthenticationFailed - case http.StatusForbidden: - return errx.ErrMailAuthorizationFailed - case http.StatusTooManyRequests: - return errx.ErrMailSendingTooFast - default: + logError := func() { log.Debug(). Int("status", resp.StatusCode). Str("code", env.Error.Code). Str("message", env.Error.Message). Msg("Graph API error") + } + + switch resp.StatusCode { + case http.StatusUnauthorized: + return errx.ErrMailAuthenticationFailed + case http.StatusForbidden: + return errx.ErrMailAuthorizationFailed + case http.StatusNotFound: + logError() + return errx.ErrMailResourceNotFound + case http.StatusTooManyRequests: + return errx.ErrMailSendingTooFast + default: + logError() return errx.ErrMailServerUnreachable } } diff --git a/internal/client/msgraph/error_test.go b/internal/client/msgraph/error_test.go new file mode 100644 index 00000000..b82ace81 --- /dev/null +++ b/internal/client/msgraph/error_test.go @@ -0,0 +1,103 @@ +package msgraph + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/warmbly/warmbly/internal/errx" + "golang.org/x/oauth2" +) + +// graphStatus boots a client whose every Graph call answers with status. The +// token is live, so nothing is refreshed and the status under test is what the +// caller classifies. +func graphStatus(t *testing.T, status int, body string) *Client { + t.Helper() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(srv.Close) + + c := &Client{Email: "sender@outlook.com"} + ctx := context.WithValue(context.Background(), oauth2.HTTPClient, &http.Client{ + Transport: rewriteTo(srv.URL), + }) + token := &oauth2.Token{AccessToken: "live", Expiry: time.Now().Add(time.Hour)} + if merr := c.Init(ctx, token, oauth2.Config{}); merr != nil { + t.Fatalf("Init: %v", merr.Message) + } + return c +} + +// rewriteTo sends every request to the test server instead of graph.microsoft.com, +// keeping the path and query the client actually built. +func rewriteTo(base string) http.RoundTripper { + target, err := url.Parse(base) + if err != nil { + panic(err) + } + return roundTripFunc(func(req *http.Request) (*http.Response, error) { + clone := req.Clone(req.Context()) + clone.URL.Scheme = target.Scheme + clone.URL.Host = target.Host + clone.Host = target.Host + return http.DefaultTransport.RoundTrip(clone) + }) +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +// The classification the backfill's folder skip rests on. 404 is the only +// status that means "this folder is not on the tenant"; every other refusal, +// a 503 above all, has to stay an unreachable server so the caller retries +// instead of writing the folder off. +func TestHandleErrorSeparatesNotFoundFromUnreachable(t *testing.T) { + cases := []struct { + name string + status int + body string + want errx.MailErrorCode + }{ + {"folder absent", http.StatusNotFound, `{"error":{"code":"ErrorItemNotFound","message":"The specified object was not found in the store."}}`, errx.MailErrorCodeNotFound}, + {"graph incident", http.StatusServiceUnavailable, `{"error":{"code":"ServiceUnavailable","message":"Server busy."}}`, errx.MailErrorCodeServerUnreachable}, + {"gateway", http.StatusBadGateway, `{}`, errx.MailErrorCodeServerUnreachable}, + {"internal", http.StatusInternalServerError, `{}`, errx.MailErrorCodeServerUnreachable}, + {"unrecognised", http.StatusTeapot, `{}`, errx.MailErrorCodeServerUnreachable}, + {"expired grant", http.StatusUnauthorized, `{}`, errx.MailErrorCodeAuthenticationFailed}, + {"missing scope", http.StatusForbidden, `{}`, errx.MailErrorCodeAuthorizationFailed}, + {"throttled", http.StatusTooManyRequests, `{}`, errx.MailErrorCodeSendingTooFast}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c := graphStatus(t, tc.status, tc.body) + _, _, err := c.ListMessagesSince(context.Background(), FolderArchive, time.Now().Add(-24*time.Hour), "", 10) + if got := mailErrorOf(t, err).Code; got != tc.want { + t.Errorf("status %d classified as %s, want %s", tc.status, got, tc.want) + } + }) + } +} + +// FetchMessage keeps its own 404 handling: a message that vanished between the +// delta item and the hydration is a skip, not an error. +func TestFetchMessageStillTreatsNotFoundAsASkip(t *testing.T) { + c := graphStatus(t, http.StatusNotFound, `{"error":{"code":"ErrorItemNotFound"}}`) + msg, err := c.FetchMessage(context.Background(), FolderInbox, "gone") + if err != nil { + t.Fatalf("FetchMessage returned %v, want a silent skip", err) + } + if msg != nil { + t.Fatalf("FetchMessage returned %+v, want nil", msg) + } +} diff --git a/internal/errx/email.go b/internal/errx/email.go index 00e23240..a30e7057 100644 --- a/internal/errx/email.go +++ b/internal/errx/email.go @@ -36,7 +36,11 @@ const ( MailErrorCodeGooglePayment MailErrorCode = "GOOGLE_PAYMENT_REQUIRED" MailErrorCodeGoogleForbidden MailErrorCode = "GOOGLE_FORBIDDEN" - MailErrorCodeServerUnreachable MailErrorCode = "SERVER_UNREACHABLE" + MailErrorCodeServerUnreachable MailErrorCode = "SERVER_UNREACHABLE" + // MailErrorCodeNotFound is the provider saying the addressed resource is + // absent, kept apart from SERVER_UNREACHABLE so "this folder does not + // exist" is never inferred from a catch-all that also covers a 503. + MailErrorCodeNotFound MailErrorCode = "RESOURCE_NOT_FOUND" MailErrorCodeUnsupported MailErrorCode = "UNSUPPORTED" MailErrorCodeInvalidCredentials MailErrorCode = "INVALID_CREDENTIALS" // e.g. invalid username or password MailErrorCodeAuthorizationFailed MailErrorCode = "AUTHORIZATION_FAILED" // e.g. imap disabled @@ -113,6 +117,7 @@ var ( return MError(MailErrorWarning, MailErrorCodeGoogleUnknown(code), message, MailErrorResolveMethodRetry) } ErrMailServerUnreachable = MError(MailErrorWarning, MailErrorCodeServerUnreachable, "The connection to the mail server could not be established. The server may be offline or blocking the connection.", MailErrorResolveMethodRetry) + ErrMailResourceNotFound = MError(MailErrorWarning, MailErrorCodeNotFound, "The mail server does not have the folder or message that was requested.", MailErrorResolveMethodRetry) ErrMailCondStoreNotSupported = MError(MailErrorCritical, MailErrorCodeUnsupported, "The mail server does not support the required CONDSTORE extension. Synchronization cannot continue.", MailErrorResolveMethodReload) ErrMailInvalidCredentials = MError( MailErrorCritical, @@ -210,6 +215,9 @@ func (e *MailError) GetUserErrorInfo() UserErrorInfo { case MailErrorCodeServerUnreachable: info.Title = "Connection Error" info.ActionRequired = "The email server is temporarily unavailable. We'll retry automatically." + case MailErrorCodeNotFound: + info.Title = "Mailbox Item Missing" + info.ActionRequired = "The folder or message is no longer on the mail server. Nothing to do; we'll skip it." case MailErrorCodeRateLimitExceeded: info.Title = "Rate Limit Exceeded" info.ActionRequired = "Your account has been temporarily limited due to unusual activity"