diff --git a/internal/app/consumer/err.go b/internal/app/consumer/err.go index ee0b4d02b..71ea9fb6c 100644 --- a/internal/app/consumer/err.go +++ b/internal/app/consumer/err.go @@ -1,11 +1,16 @@ package jobs import ( + "errors" + "github.com/google/uuid" "github.com/warmbly/warmbly/internal/observability/errs" ) func CaptureError(userID, emailID uuid.UUID, err error) { + if errors.Is(err, ErrSyncArrivalPending) { + return + } errs.CaptureException(err, errs.Tag("user_id", userID.String()), errs.Tag("email_id", emailID.String()), diff --git a/internal/app/consumer/sync_truth_test.go b/internal/app/consumer/sync_truth_test.go index 148147e30..6c0d1f6a8 100644 --- a/internal/app/consumer/sync_truth_test.go +++ b/internal/app/consumer/sync_truth_test.go @@ -1,17 +1,46 @@ package jobs import ( + "bytes" "context" "errors" + "fmt" + "log" + "os" + "strings" "testing" "time" "github.com/google/uuid" "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/observability/errs" "github.com/warmbly/warmbly/internal/repository" ) +func TestArrivalWaitStillRetriesWithoutCapturingAnException(t *testing.T) { + if err := errs.Init(errs.Config{Service: "consumer", Environment: "test"}); err != nil { + t.Fatal(err) + } + var output bytes.Buffer + log.SetOutput(&output) + t.Cleanup(func() { log.SetOutput(os.Stderr) }) + user, mailbox, id := uuid.New(), uuid.New(), uuid.New() + s := &JobsService{UniboxRepository: &syncTruthLookup{}, ArrivalOutbox: &syncTruthOutbox{user: user, mailbox: mailbox, id: id, pending: true}} + err := s.HandleUpdateEmail(t.Context(), &models.JobEventEmailUpdate{UserID: user, EmailID: mailbox, ID: id}) + if !errors.Is(err, ErrSyncArrivalPending) { + t.Fatalf("arrival wait acknowledged instead of retried: %v", err) + } + CaptureError(user, mailbox, fmt.Errorf("wrapped: %w", err)) + if output.Len() != 0 { + t.Fatalf("expected wait captured as an exception: %s", output.String()) + } + CaptureError(user, mailbox, errors.New("actual repository failure")) + if !strings.Contains(output.String(), "actual repository failure") { + t.Fatal("unexpected errors are no longer reported") + } +} + func TestSyncTruthProgressRelaysDoNotResolveTransportWarnings(t *testing.T) { previous := time.Now().Add(-time.Hour) fresh := previous.Add(time.Minute) diff --git a/internal/app/worker/wmail/graph_recovery_test.go b/internal/app/worker/wmail/graph_recovery_test.go index 9c453ce86..f060ee125 100644 --- a/internal/app/worker/wmail/graph_recovery_test.go +++ b/internal/app/worker/wmail/graph_recovery_test.go @@ -17,12 +17,17 @@ import ( ) type graphRecoveryProvider struct { - snapshot map[string][]any - page2 map[string][]any - messages map[string]map[string]any - expired string - seen bool - requests []string + snapshot map[string][]any + page2 map[string][]any + messages map[string]map[string]any + translations map[string]string + messageErrors map[string]int + resolvedID string + resolveError int + resolveMalformed bool + expired string + seen bool + requests []string } func (g *graphRecoveryProvider) serve(t *testing.T) *httptest.Server { @@ -35,6 +40,23 @@ func (g *graphRecoveryProvider) serve(t *testing.T) *httptest.Server { path := strings.TrimPrefix(r.URL.Path, "/v1.0/me/") g.requests = append(g.requests, path+"?"+r.URL.RawQuery) switch { + case path == "messages": + if r.URL.Query().Get("$filter") != "internetMessageId eq ''" { + t.Error("RFC lookup lost its exact stored identity") + } + if g.resolveError != 0 { + w.WriteHeader(g.resolveError) + _, _ = w.Write([]byte(`{"error":{"code":"ErrorServerBusy"}}`)) + return + } + value := []any{} + if g.resolveMalformed { + value = append(value, map[string]any{}) + } + if g.resolvedID != "" { + value = append(value, map[string]any{"id": g.resolvedID}) + } + _ = json.NewEncoder(w).Encode(map[string]any{"value": value}) case path == "translateExchangeIds": var body struct { IDs []string `json:"inputIds"` @@ -49,11 +71,19 @@ func (g *graphRecoveryProvider) serve(t *testing.T) *httptest.Server { if body.Target == "restId" { target = "regular-" + strings.TrimPrefix(id, "stable-") } + if translated, ok := g.translations[id]; ok { + target = translated + } value = append(value, map[string]any{"sourceId": id, "targetId": target}) } _ = json.NewEncoder(w).Encode(map[string]any{"value": value}) case strings.HasPrefix(path, "messages/"): id := strings.TrimPrefix(path, "messages/") + if status := g.messageErrors[id]; status != 0 { + w.WriteHeader(status) + _, _ = w.Write([]byte(`{"error":{"code":"ErrorServerBusy"}}`)) + return + } msg := g.messages[id] if msg == nil { w.WriteHeader(http.StatusNotFound) @@ -102,6 +132,125 @@ func (g *graphRecoveryProvider) serve(t *testing.T) *httptest.Server { return srv } +func TestGraphIdentityUpgradeRecoversPartialConversions(t *testing.T) { + for _, tc := range []struct { + name string + message map[string]any + status int + resolvedID string + resolveError int + wantErr bool + wantRemoved bool + }{ + {"live message", graphMessageJSON("stable-recovered"), 0, "", 0, false, false}, + {"moved message", nil, 0, "stable-recovered", 0, false, false}, + {"confirmed missing", nil, 0, "", 0, false, true}, + {"provider unavailable", nil, http.StatusServiceUnavailable, "", 0, true, false}, + {"RFC lookup unavailable", nil, 0, "", http.StatusServiceUnavailable, true, false}, + {"empty message identity", map[string]any{}, 0, "", 0, true, false}, + } { + t.Run(tc.name, func(t *testing.T) { + g := &graphRecoveryProvider{translations: map[string]string{"regular-recover": ""}, messages: map[string]map[string]any{"regular-recover": tc.message}, messageErrors: map[string]int{"regular-recover": tc.status}, resolvedID: tc.resolvedID, resolveError: tc.resolveError} + var events []captured + var states []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &states) + w.GraphData.Client.SetImmutableIDMode(true) + id := uuid.New() + w.EmailMessageMapRepository = &recoveryMessageMap{data: map[string]repository.EmailMessageData{"regular-recover": {UserID: w.UserID.String(), EmailID: w.ID.String(), MessageID: "regular-recover", ID: id.String(), ThreadID: "owned"}}} + w.SyncContext = &graphRecoveryContext{recoveryRows: recoveryRows{rows: []repository.ProviderFolderMessage{{ID: id, ProviderID: "regular-recover", MessageID: "", ProviderFolder: models.FolderInbox}}}} + done, err := w.graphUpgradeIDs(t.Context()) + if (err != nil) != tc.wantErr || done == tc.wantErr { + t.Fatalf("done=%v err=%v", done, err) + } + cur := w.tracker.folder(graphIdentityCursor) + if tc.wantErr { + if cur.Next != "" || cur.Done || len(events) != 0 { + t.Fatalf("failed lookup advanced or deleted mail: cursor=%+v events=%v", cur, events) + } + return + } + if cur.Next != id.String() || !cur.Done || len(events) != 1 { + t.Fatalf("upgrade failed to checkpoint: cursor=%+v events=%v", cur, events) + } + if tc.wantRemoved { + if events[0].eventType != models.JobEventTypeRemoveEmail || events[0].body.(*models.JobEventRemoveEmail).ID != id { + t.Fatal("confirmed missing message not removed with original identity") + } + } else { + update := events[0].body.(*models.JobEventFolderUpdate) + if update.ID != id || update.ProviderID != "stable-recovered" || !update.Relayed { + t.Fatalf("recovered identity changed the conversation: %+v", update) + } + } + }) + } +} + +func TestGraphIdentityUpgradeResumesAfterTransientFailure(t *testing.T) { + first := uuid.MustParse("00000000-0000-4000-8000-000000000001") + second := uuid.MustParse("00000000-0000-4000-8000-000000000002") + g := &graphRecoveryProvider{translations: map[string]string{"regular-recover": ""}, messages: map[string]map[string]any{}, messageErrors: map[string]int{"regular-recover": http.StatusServiceUnavailable}} + srv := g.serve(t) + var events []captured + var states []models.SyncState + w := newGraphTestMail(t, srv, &events, &states) + w.GraphData.Client.SetImmutableIDMode(true) + maps := &recoveryMessageMap{data: map[string]repository.EmailMessageData{}} + for key, id := range map[string]uuid.UUID{"regular-first": first, "regular-recover": second} { + maps.data[key] = repository.EmailMessageData{UserID: w.UserID.String(), EmailID: w.ID.String(), MessageID: key, ID: id.String(), ThreadID: "owned"} + } + w.EmailMessageMapRepository = maps + w.SyncContext = &graphRecoveryContext{recoveryRows: recoveryRows{rows: []repository.ProviderFolderMessage{ + {ID: first, ProviderID: "regular-first", ProviderFolder: models.FolderInbox}, + {ID: second, ProviderID: "regular-recover", MessageID: "", ProviderFolder: models.FolderInbox}, + }}} + if done, err := w.graphUpgradeIDs(t.Context()); err == nil || done { + t.Fatalf("transient conversion marked upgrade done: done=%v err=%v", done, err) + } + if cur := w.tracker.folder(graphIdentityCursor); cur.Next != first.String() || cur.Done || len(events) != 1 { + t.Fatalf("lost successful prefix or advanced past failure: cursor=%+v events=%v", cur, events) + } + g.messageErrors = nil + g.resolvedID = "stable-recovered" + reloaded := newGraphTestMail(t, srv, &events, &states) + reloaded.ID, reloaded.UserID = w.ID, w.UserID + reloaded.GraphData.Client.SetImmutableIDMode(true) + reloaded.EmailMessageMapRepository, reloaded.SyncContext = maps, w.SyncContext + reloaded.tracker = newSyncTracker(graphReloadState(t, w.tracker.state), func(models.SyncState) error { return nil }) + if done, err := reloaded.graphUpgradeIDs(t.Context()); err != nil || !done { + t.Fatalf("retry failed to recover: done=%v err=%v", done, err) + } + if len(events) != 2 || maps.data["stable-first"].ID != first.String() || maps.data["stable-recovered"].ID != second.String() { + t.Fatalf("retry changed identity or replayed prefix: events=%v maps=%v", events, maps.data) + } +} + +func TestGraphIdentityUpgradeWithoutRFCIdentityNeverDeletesOnLegacy404(t *testing.T) { + g := &graphRecoveryProvider{translations: map[string]string{"regular-missing": ""}, messages: map[string]map[string]any{}} + var events []captured + var states []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &states) + w.GraphData.Client.SetImmutableIDMode(true) + w.SyncContext = &graphRecoveryContext{recoveryRows: recoveryRows{rows: []repository.ProviderFolderMessage{{ID: uuid.New(), ProviderID: "regular-missing"}}}} + if done, err := w.graphUpgradeIDs(t.Context()); !errors.Is(err, repository.ErrSyncContextUnsupported) || done { + t.Fatalf("missing stable identity did not retain legacy mode: done=%v err=%v", done, err) + } + if cur := w.tracker.folder(graphIdentityCursor); cur.Next != "" || cur.Done || len(events) != 0 { + t.Fatalf("legacy 404 deleted or checkpointed mail: cursor=%+v events=%v", cur, events) + } +} + +func TestGraphRFCIdentityLookupRejectsMalformedResults(t *testing.T) { + g := &graphRecoveryProvider{resolveMalformed: true} + var events []captured + var states []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &states) + w.GraphData.Client.SetImmutableIDMode(true) + if _, err := w.GraphData.Client.ResolveMessageID(t.Context(), ""); err == nil { + t.Fatal("malformed lookup reported definitive absence") + } +} + type graphRecoveryContext struct{ recoveryRows } func (*graphRecoveryContext) IsOwnConversation(_ context.Context, _, _ uuid.UUID, _ []string, thread string) (bool, error) { diff --git a/internal/app/worker/wmail/sync_graph_reconcile.go b/internal/app/worker/wmail/sync_graph_reconcile.go index 45d4af032..02524619e 100644 --- a/internal/app/worker/wmail/sync_graph_reconcile.go +++ b/internal/app/worker/wmail/sync_graph_reconcile.go @@ -191,13 +191,43 @@ func (w *WMail) graphUpgradeIDs(ctx context.Context) (bool, error) { var converted map[string]string if len(ids) > 0 { converted, err = w.GraphData.Client.ImmutableMessageIDs(ctx, ids) - if err != nil { + if err != nil && !errors.Is(err, msgraph.ErrIncompleteIDConversion) { return false, err } } cur := w.tracker.folder(graphIdentityCursor) for _, row := range rows { id := converted[row.ProviderID] + if id == "" { + // A per-item conversion failure is not proof that the message was deleted. + full, err := w.GraphData.Client.FetchMessage(ctx, "", row.ProviderID) + if err != nil { + return false, err + } + if full == nil { + // Regular IDs also disappear on moves; resolve the stable RFC Message-ID first. + if row.MessageID == "" { + return false, repository.ErrSyncContextUnsupported + } + id, err = w.GraphData.Client.ResolveMessageID(ctx, row.MessageID) + if err != nil { + return false, err + } + } else { + id = full.ID + if id == "" { + return false, msgraph.ErrIncompleteIDConversion + } + } + if id == "" { + if err := w.onEvent(models.JobEventTypeRemoveEmail, &models.JobEventRemoveEmail{UserID: w.UserID, EmailID: w.ID, ID: row.ID}); err != nil { + return false, err + } + cur.Next = row.ID.String() + w.tracker.setFolder(graphIdentityCursor, cur) + continue + } + } known, err := w.EmailMessageMapRepository.Get(ctx, w.UserID, w.ID, row.ProviderID) if err != nil { return false, err diff --git a/internal/client/msgraph/actions.go b/internal/client/msgraph/actions.go index bffaf63ed..75ebc643e 100644 --- a/internal/client/msgraph/actions.go +++ b/internal/client/msgraph/actions.go @@ -167,6 +167,9 @@ func (c *Client) ResolveMessageID(ctx context.Context, internetMessageID string) if len(resp.Value) == 0 { return "", nil } + if resp.Value[0].ID == "" { + return "", errors.New("graph: message lookup returned an empty id") + } return resp.Value[0].ID, nil } diff --git a/internal/client/msgraph/sync_identity.go b/internal/client/msgraph/sync_identity.go index 7c156797a..9f812de17 100644 --- a/internal/client/msgraph/sync_identity.go +++ b/internal/client/msgraph/sync_identity.go @@ -2,13 +2,15 @@ package msgraph import ( "context" - "fmt" + "errors" "net/http" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" ) +var ErrIncompleteIDConversion = errors.New("graph: incomplete immutable id conversion") + func (c *Client) ImmutableIDMode() bool { c.idMu.RLock() defer c.idMu.RUnlock() @@ -50,7 +52,7 @@ func (c *Client) translateIDs(ctx context.Context, ids []string, source, target } for _, id := range ids { if converted[id] == "" { - return nil, fmt.Errorf("graph: incomplete immutable id conversion") + return converted, ErrIncompleteIDConversion } } return converted, nil diff --git a/internal/repository/http_graph_upgrade_test.go b/internal/repository/http_graph_upgrade_test.go index 27b0efca4..7ef0b0eb8 100644 --- a/internal/repository/http_graph_upgrade_test.go +++ b/internal/repository/http_graph_upgrade_test.go @@ -1,6 +1,7 @@ package repository import ( + "encoding/json" "errors" "net/http" "net/http/httptest" @@ -9,6 +10,26 @@ import ( "github.com/google/uuid" ) +func TestProviderMessagesCarriesRFCIdentityAndAcceptsLegacyRows(t *testing.T) { + for _, messageID := range []string{"", ""} { + t.Run(messageID, func(t *testing.T) { + user, mailbox, id := uuid.New(), uuid.New(), uuid.New() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("user_id") != user.String() || r.URL.Query().Get("email_id") != mailbox.String() { + t.Error("provider enumeration lost mailbox scope") + } + _ = json.NewEncoder(w).Encode(map[string]any{"messages": []ProviderFolderMessage{{ID: id, ProviderID: "regular", MessageID: messageID}}}) + })) + t.Cleanup(srv.Close) + repo := &httpSyncContextRepository{baseURL: srv.URL, client: srv.Client()} + rows, err := repo.ListProviderMessages(t.Context(), user, mailbox, nil, 10) + if err != nil || len(rows) != 1 || rows[0].MessageID != messageID { + t.Fatalf("rows=%v err=%v", rows, err) + } + }) + } +} + func TestProviderMessagesOldBackendIsTypedUnsupported(t *testing.T) { for _, status := range []int{http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusServiceUnavailable, http.StatusForbidden} { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(status) })) diff --git a/internal/repository/http_sync_context.go b/internal/repository/http_sync_context.go index 4434fda40..730e7458e 100644 --- a/internal/repository/http_sync_context.go +++ b/internal/repository/http_sync_context.go @@ -32,6 +32,7 @@ type StoredFolderMessage struct { type ProviderFolderMessage struct { ID uuid.UUID `json:"id"` ProviderID string `json:"provider_id"` + MessageID string `json:"message_id,omitempty"` ProviderFolder string `json:"provider_folder"` InternalDate time.Time `json:"internal_date"` Flags []string `json:"flags,omitempty"` diff --git a/internal/repository/pg_email_sync_state.go b/internal/repository/pg_email_sync_state.go index 1d1bca035..309f235db 100644 --- a/internal/repository/pg_email_sync_state.go +++ b/internal/repository/pg_email_sync_state.go @@ -38,7 +38,7 @@ type EmailSyncStateRepository interface { } func (r *pgEmailSyncStateRepository) ListProviderMessages(ctx context.Context, userID, emailID uuid.UUID, after *uuid.UUID, limit int) ([]ProviderFolderMessage, error) { - const q = `SELECT id, gmail_id, provider_folder, internal_date, flags FROM unibox_emails + const q = `SELECT id, gmail_id, message_id, provider_folder, internal_date, flags FROM unibox_emails WHERE user_id = $1 AND email_id = $2 AND gmail_id <> '' AND ($3::uuid IS NULL OR id > $3) ORDER BY id LIMIT $4` rows, err := r.db.Query(ctx, q, userID, emailID, after, limit) @@ -49,7 +49,7 @@ func (r *pgEmailSyncStateRepository) ListProviderMessages(ctx context.Context, u out := make([]ProviderFolderMessage, 0) for rows.Next() { var m ProviderFolderMessage - if err := rows.Scan(&m.ID, &m.ProviderID, &m.ProviderFolder, &m.InternalDate, &m.Flags); err != nil { + if err := rows.Scan(&m.ID, &m.ProviderID, &m.MessageID, &m.ProviderFolder, &m.InternalDate, &m.Flags); err != nil { return nil, err } out = append(out, m)