feat: recover partial Outlook immutable ID upgrades without losing moved conversations and retain arrival retries without exception floods

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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()),

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)

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 '<stored@fake.test>'" {
+				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: "<stored@fake.test>", 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: "<stored@fake.test>", 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(), "<stored@fake.test>"); 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) {

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

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
 }
 

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

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{"<stored@fake.test>", ""} {
+		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) }))

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"`

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)