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"