From cbfd39a0527d7045b12dcba47574d8a1b9eb13f4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 9 Oct 2026 12:34:38 +0000 Subject: [PATCH] feat: preserve truthful cross-provider sync success and backlog evidence, retry durable pending-arrival updates without reading unread mail, and render factual sync status --- internal/app/consumer/event_flags_update.go | 4 +- internal/app/consumer/event_sync_state.go | 11 +- .../app/consumer/event_sync_state_test.go | 5 +- internal/app/consumer/event_update_email.go | 29 +- internal/app/consumer/sync_truth_live_test.go | 193 ++++++++++ internal/app/consumer/sync_truth_test.go | 167 ++++++++ .../app/consumer/unibox_seen_sync_test.go | 4 + internal/app/worker/wmail/admit.go | 18 +- internal/app/worker/wmail/start_sync.go | 11 +- internal/app/worker/wmail/sync_google.go | 12 +- internal/app/worker/wmail/sync_graph.go | 8 +- internal/app/worker/wmail/sync_imap.go | 8 +- internal/app/worker/wmail/sync_imap_flags.go | 3 +- internal/app/worker/wmail/sync_state.go | 14 +- internal/app/worker/wmail/sync_truth_test.go | 355 ++++++++++++++++++ internal/client/goog/history.go | 23 +- internal/client/msgraph/delta.go | 33 +- internal/repository/pg_unibox.go | 30 +- .../app/emails/SyncStatusCard.test.tsx | 70 ++++ .../components/app/emails/SyncStatusCard.tsx | 47 ++- web/src/lib/api/hooks/app/emails/useSync.ts | 4 +- .../lib/api/models/app/emails/SyncState.ts | 11 + 22 files changed, 995 insertions(+), 65 deletions(-) create mode 100644 internal/app/consumer/sync_truth_live_test.go create mode 100644 internal/app/consumer/sync_truth_test.go create mode 100644 internal/app/worker/wmail/sync_truth_test.go create mode 100644 web/src/components/app/emails/SyncStatusCard.test.tsx diff --git a/internal/app/consumer/event_flags_update.go b/internal/app/consumer/event_flags_update.go index a2fe5f5a8..39cd3f73a 100644 --- a/internal/app/consumer/event_flags_update.go +++ b/internal/app/consumer/event_flags_update.go @@ -38,7 +38,7 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag } } - email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) { + email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) { for _, flag := range e.Flags { if !slices.Contains(message.Flags, flag) { message.Flags = append(message.Flags, flag) @@ -155,7 +155,7 @@ func warmupTokenFromFlags(flags []string) string { } func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventFlags) error { - email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) { + email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) { message.Flags = slices.DeleteFunc(message.Flags, func(flag string) bool { return slices.Contains(e.Flags, flag) }) if models.SeenFromFlags(e.Flags) { message.Seen = false diff --git a/internal/app/consumer/event_sync_state.go b/internal/app/consumer/event_sync_state.go index dd6565523..17d381df4 100644 --- a/internal/app/consumer/event_sync_state.go +++ b/internal/app/consumer/event_sync_state.go @@ -34,12 +34,11 @@ func (s *JobsService) HandleSyncState(ctx context.Context, e *models.JobEventSyn return err } - // A relayed state means a pass completed: the worker reached the server, - // listed its folders and finished the tick. That is the only signal we - // get that an outage is over, and without it a five-minute blip left a - // red "needs attention" on the mailbox for good, because nothing but a - // credential reconnect ever resolved an error row. - s.resolveTransientMailErrors(ctx, e.EmailID, e.State.LastSyncedAt) + // Progress relays are not recovery evidence; require a newer successful check. + if e.State.LastSyncedAt != nil && e.State.Deferred == 0 && e.State.BackfillCursor.GoogleRecovery == nil && + (prev == nil || prev.LastSyncedAt == nil || e.State.LastSyncedAt.After(*prev.LastSyncedAt)) { + s.resolveTransientMailErrors(ctx, e.EmailID, e.State.LastSyncedAt) + } if s.StreamingPublisher == nil { return nil diff --git a/internal/app/consumer/event_sync_state_test.go b/internal/app/consumer/event_sync_state_test.go index 6218eacb4..9d077256d 100644 --- a/internal/app/consumer/event_sync_state_test.go +++ b/internal/app/consumer/event_sync_state_test.go @@ -14,11 +14,12 @@ import ( // stubSyncStateRepo accepts what the worker relays. type stubSyncStateRepo struct { repository.EmailSyncStateRepository - put *models.SyncState + put *models.SyncState + saved *models.SyncState } func (s *stubSyncStateRepo) Get(context.Context, uuid.UUID) (*models.SyncState, error) { - return nil, nil + return s.saved, nil } func (s *stubSyncStateRepo) Put(_ context.Context, _, _ uuid.UUID, st *models.SyncState) error { diff --git a/internal/app/consumer/event_update_email.go b/internal/app/consumer/event_update_email.go index 128545cf6..216995de3 100644 --- a/internal/app/consumer/event_update_email.go +++ b/internal/app/consumer/event_update_email.go @@ -13,7 +13,7 @@ import ( ) func (s *JobsService) HandleUpdateEmail(ctx context.Context, e *models.JobEventEmailUpdate) error { - email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) { + email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) { token := warmupTokenFromMessage(message) message.Flags = append([]string{}, e.Flags...) if token != "" { @@ -98,7 +98,7 @@ func (s *JobsService) HandleFolderUpdate(ctx context.Context, e *models.JobEvent if e.Relayed { return s.applyRelayedFolder(ctx, e) } - email, err := s.emailForSyncUpdate(ctx, e.UserID, e.ID, func(message *models.EmailMessageStoreData) { + email, err := s.emailForSyncUpdate(ctx, e.UserID, e.EmailID, e.ID, func(message *models.EmailMessageStoreData) { // A pending row is not visible yet, so nothing has filed it locally. message.Folder = e.Folder }) @@ -144,16 +144,33 @@ func (s *JobsService) applyRelayedFolder(ctx context.Context, e *models.JobEvent } // emailForSyncUpdate rechecks visible mail if verification won the pending-row lock. -func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, id uuid.UUID, updatePending func(*models.EmailMessageStoreData)) (*models.EmailMessageStoreData, error) { - message, err := s.UniboxRepository.GetByID(ctx, userID, id) +var ErrSyncArrivalPending = errors.New("sync arrival pending delivery") + +func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, emailID, id uuid.UUID, updatePending func(*models.EmailMessageStoreData)) (*models.EmailMessageStoreData, error) { + message, err := s.UniboxRepository.GetForSync(ctx, userID, emailID, id) if !errors.Is(err, repository.ErrEmailNotFound) { return message, err } - updated, err := s.UniboxRepository.UpdatePendingEmail(ctx, userID, id, updatePending) + updated, err := s.UniboxRepository.UpdatePendingEmail(ctx, userID, id, func(message *models.EmailMessageStoreData) { + if message.EmailID == emailID { + updatePending(message) + } + }) if err != nil || updated { return nil, err } - message, err = s.UniboxRepository.GetByID(ctx, userID, id) + message, err = s.UniboxRepository.GetForSync(ctx, userID, emailID, id) + if errors.Is(err, repository.ErrEmailNotFound) && s.ArrivalOutbox != nil { + pending, lookupErr := s.ArrivalOutbox.HasPendingArrival(ctx, userID, emailID, id) + if lookupErr != nil { + return nil, lookupErr + } + if pending { + return nil, ErrSyncArrivalPending + } + // Delivery may have finished between the visible-row read and the marker read. + message, err = s.UniboxRepository.GetForSync(ctx, userID, emailID, id) + } if errors.Is(err, repository.ErrEmailNotFound) { return nil, nil } diff --git a/internal/app/consumer/sync_truth_live_test.go b/internal/app/consumer/sync_truth_live_test.go new file mode 100644 index 000000000..299a548da --- /dev/null +++ b/internal/app/consumer/sync_truth_live_test.go @@ -0,0 +1,193 @@ +package jobs + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/cipher" + "github.com/warmbly/warmbly/internal/infrastructure/encryptedkeys" + "github.com/warmbly/warmbly/internal/infrastructure/kms" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type failSyncArrivalCreate struct { + repository.UniboxRepository + fail bool +} + +func (r *failSyncArrivalCreate) CreateEntry(ctx context.Context, user uuid.UUID, message *models.EmailMessageStoreData) error { + if r.fail { + r.fail = false + return errors.New("transient arrival write failure") + } + return r.UniboxRepository.CreateEntry(ctx, user, message) +} + +func TestLiveSyncTruthPendingArrivalRetainsProviderUpdatesAcrossReload(t *testing.T) { + for _, kind := range []string{"seen", "folder", "full update"} { + t.Run(kind, func(t *testing.T) { + s, d := liveWarmupService(t) + f := newWarmupFixture(t, d) + ctx := t.Context() + key, err := kms.NewLocal(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + cipherService := cipher.NewService(key, nil, encryptedkeys.NewPostgres(d)) + r := repository.NewDurableEmailMessageMapRepository(d, cipherService) + s.ArrivalOutbox = r + s.UniboxRepository = &failSyncArrivalCreate{UniboxRepository: s.UniboxRepository, fail: true} + e := f.arrival("<"+uuid.NewString()+"@test.local>", nil) + e.Message.Subject = "ordinary customer question" + mapping := repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), ID: e.Message.ID.String(), MessageID: e.Message.MessageID} + if err := r.AdmitArrival(ctx, mapping, &repository.PendingArrival{Arrival: e}); err != nil { + t.Fatal(err) + } + s.InitEvents() + if err := r.DeliverArrivals(ctx, s.deliverArrivalEvent); err == nil { + t.Fatal("expected transient arrival write failure") + } + apply := func(service *JobsService) error { + switch kind { + case "seen": + return service.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}}) + case "folder": + return service.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: models.FolderArchive}) + default: + return service.HandleUpdateEmail(ctx, &models.JobEventEmailUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}, Folder: models.FolderArchive, FolderPath: "Archive", UID: 42, Mailbox: 7}) + } + } + if err := apply(s); !errors.Is(err, ErrSyncArrivalPending) { + t.Fatalf("update was acknowledged ahead of failed arrival: %v", err) + } + if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET retry_at=now() WHERE id=$1`, e.Message.ID); err != nil { + t.Fatal(err) + } + fresh, _ := liveWarmupService(t) + fresh.ArrivalOutbox = repository.NewDurableEmailMessageMapRepository(d, cipherService) + fresh.InitEvents() + if err := fresh.ArrivalOutbox.DeliverArrivals(ctx, fresh.deliverArrivalEvent); err != nil { + t.Fatal(err) + } + if err := apply(fresh); err != nil { + t.Fatal(err) + } + message, err := fresh.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID) + if err != nil { + t.Fatal(err) + } + if kind != "folder" && !message.Seen { + t.Fatal("provider Seen update lost after original-ID arrival replay") + } + if kind != "seen" && message.Folder != models.FolderArchive { + t.Fatalf("provider folder update lost: %q", message.Folder) + } + if kind == "full update" && (message.UID != 42 || message.Mailbox != 7 || message.FolderPath != "Archive") { + t.Fatalf("provider handles lost: %+v", message) + } + if _, err := d.Exec(ctx, `DELETE FROM email_accounts WHERE id=$1`, e.Message.EmailID); err != nil { + t.Fatal(err) + } + if err := apply(fresh); err != nil { + t.Fatalf("deleted mailbox update retried forever: %v", err) + } + }) + } +} + +func TestLiveSyncTruthBackgroundReadsKeepUnreadAndStayOwnerScoped(t *testing.T) { + s, d := liveWarmupService(t) + f := newWarmupFixture(t, d) + ctx := t.Context() + e := f.arrival("<"+uuid.NewString()+"@test.local>", nil) + if err := s.UniboxRepository.CreateEntry(ctx, e.UserID, e.Message); err != nil { + t.Fatal(err) + } + assertUnread := func() { + t.Helper() + var seen bool + if err := d.QueryRow(ctx, `SELECT seen FROM unibox_emails WHERE id=$1`, e.Message.ID).Scan(&seen); err != nil || seen { + t.Fatalf("background work marked unread mail seen: %v %v", seen, err) + } + } + for _, owner := range []uuid.UUID{f.senderUser, uuid.New()} { + if _, err := s.UniboxRepository.GetForSync(ctx, owner, e.Message.EmailID, e.Message.ID); !errors.Is(err, repository.ErrEmailNotFound) { + t.Fatalf("unowned sync lookup returned a row: %v", err) + } + } + for _, mailbox := range []uuid.UUID{f.sender, uuid.New(), uuid.Nil} { + if _, err := s.UniboxRepository.GetForSync(ctx, e.UserID, mailbox, e.Message.ID); !errors.Is(err, repository.ErrEmailNotFound) { + t.Fatalf("wrong-mailbox sync lookup returned a row: %v", err) + } + } + assertUnread() + if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagFlagged}}); err != nil { + t.Fatal(err) + } + assertUnread() + for _, folder := range []string{models.FolderInbox, models.FolderArchive, models.FolderArchive} { + if err := s.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: folder}); err != nil { + t.Fatal(err) + } + assertUnread() + } + if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}}); err != nil { + t.Fatal(err) + } + message, err := s.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID) + if err != nil || message == nil || !message.Seen { + t.Fatalf("real provider Seen flag did not mark read: %v", err) + } + if err := s.UniboxRepository.MarkSeen(ctx, e.UserID, e.Message.ID, false); err != nil { + t.Fatal(err) + } + assertUnread() + message, err = s.UniboxRepository.GetByID(ctx, e.UserID, e.Message.ID) + if err != nil || message == nil || !message.Seen { + t.Fatalf("explicit user view no longer marks read: %v", err) + } +} + +func TestLiveSyncTruthLegacyArrivalHasNoPendingGuard(t *testing.T) { + s, d := liveWarmupService(t) + f := newWarmupFixture(t, d) + ctx := t.Context() + key, err := kms.NewLocal(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + r := repository.NewDurableEmailMessageMapRepository(d, cipher.NewService(key, nil, encryptedkeys.NewPostgres(d))) + s.ArrivalOutbox = r + s.UniboxRepository = &failSyncArrivalCreate{UniboxRepository: s.UniboxRepository, fail: true} + e := f.arrival("<"+uuid.NewString()+"@test.local>", nil) + e.Message.Subject = "ordinary legacy customer question" + // Legacy workers admit only the map, then publish NEW_EMAIL separately. + if err := r.Add(ctx, repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), ID: e.Message.ID.String(), MessageID: e.Message.MessageID}); err != nil { + t.Fatal(err) + } + if err := s.HandleNewEmail(ctx, e); err == nil { + t.Fatal("expected legacy arrival creation failure") + } + if pending, err := r.HasPendingArrival(ctx, e.UserID, e.Message.EmailID, e.Message.ID); err != nil || pending { + t.Fatalf("legacy map unexpectedly acquired a durable arrival marker: %v %v", pending, err) + } + if err := s.HandleFlagsAdd(ctx, &models.JobEventFlags{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{models.FlagSeen}}); err != nil { + t.Fatalf("legacy map alone inferred a pending arrival: %v", err) + } + if err := s.HandleFolderUpdate(ctx, &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: models.FolderArchive}); err != nil { + t.Fatalf("legacy map alone inferred a pending arrival: %v", err) + } + if err := s.HandleNewEmail(ctx, e); err != nil { + t.Fatal(err) + } + message, err := s.UniboxRepository.GetForSync(ctx, e.UserID, e.Message.EmailID, e.Message.ID) + if err != nil { + t.Fatal(err) + } + if message.Seen || message.Folder != models.FolderInbox { + t.Fatalf("legacy update ordering must not be reported as protected: %+v", message) + } +} diff --git a/internal/app/consumer/sync_truth_test.go b/internal/app/consumer/sync_truth_test.go new file mode 100644 index 000000000..74abc19da --- /dev/null +++ b/internal/app/consumer/sync_truth_test.go @@ -0,0 +1,167 @@ +package jobs + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +func TestSyncTruthProgressRelaysDoNotResolveTransportWarnings(t *testing.T) { + previous := time.Now().Add(-time.Hour) + fresh := previous.Add(time.Minute) + for _, tc := range []struct { + name string + stamp *time.Time + deferred int + recovery *models.GoogleHistoryRecovery + resolved int + }{ + {"heartbeat", &previous, 0, nil, 0}, + {"older redelivery", nil, 0, nil, 0}, + {"deferred", &fresh, 5, nil, 0}, + {"recovering", &fresh, 0, &models.GoogleHistoryRecovery{HistoryID: 100}, 0}, + {"successful catch-up", &fresh, 0, nil, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + errRepo := &stubErrorRepo{} + repo := &stubSyncStateRepo{saved: &models.SyncState{LastSyncedAt: &previous}} + s := &JobsService{EmailSyncStateRepository: repo, EmailAccountErrorRepository: errRepo} + if err := s.HandleSyncState(t.Context(), &models.JobEventSyncState{UserID: uuid.New(), EmailID: uuid.New(), State: models.SyncState{BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: tc.stamp, Deferred: tc.deferred, BackfillCursor: models.SyncCursor{GoogleRecovery: tc.recovery}}}); err != nil { + t.Fatal(err) + } + if repo.put == nil || errRepo.calls != tc.resolved { + t.Fatalf("state publication/recovery outcome: saved=%v resolved=%d", repo.put != nil, errRepo.calls) + } + }) + } +} + +type syncTruthLookup struct { + repository.UniboxRepository + message *models.EmailMessageStoreData + reads int + lookups int + appearOn int +} + +func (r *syncTruthLookup) GetByID(context.Context, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) { + r.reads++ + if r.message == nil { + return nil, repository.ErrEmailNotFound + } + r.message.Seen = true + return r.message, nil +} + +func (r *syncTruthLookup) GetForSync(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) { + r.lookups++ + if r.message == nil || (r.appearOn > 0 && r.lookups < r.appearOn) { + return nil, repository.ErrEmailNotFound + } + return r.message, nil +} + +func (*syncTruthLookup) UpdatePendingEmail(context.Context, uuid.UUID, uuid.UUID, func(*models.EmailMessageStoreData)) (bool, error) { + return false, nil +} + +func (r *syncTruthLookup) UpdateEntry(_ context.Context, _, _, _ uuid.UUID, update *repository.UpdateUniboxEntry) error { + if update.Seen != nil { + r.message.Seen = *update.Seen + } + return nil +} + +type syncTruthOutbox struct { + repository.ArrivalOutbox + user, mailbox, id uuid.UUID + pending bool + err error +} + +func (r *syncTruthOutbox) HasPendingArrival(_ context.Context, user, mailbox, id uuid.UUID) (bool, error) { + if user != r.user || mailbox != r.mailbox || id != r.id { + return false, nil + } + return r.pending, r.err +} + +func TestSyncTruthBackgroundUpdatesNeverViewUnreadMail(t *testing.T) { + for _, kind := range []string{"flagged", "folder", "same folder"} { + t.Run(kind, func(t *testing.T) { + r := &syncTruthLookup{message: &models.EmailMessageStoreData{Seen: false, Folder: models.FolderInbox, ProviderFolder: models.FolderInbox}} + s := &JobsService{UniboxRepository: r, EmailRepository: warmupInboxEmailRepo{}} + user, mailbox, id := uuid.New(), uuid.New(), uuid.New() + var err error + if kind == "flagged" { + err = s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: user, EmailID: mailbox, ID: id, Flags: []string{models.FlagFlagged}}) + } else { + folder := models.FolderInbox + if kind == "folder" { + folder = models.FolderArchive + } + err = s.HandleFolderUpdate(t.Context(), &models.JobEventFolderUpdate{UserID: user, EmailID: mailbox, ID: id, Folder: folder}) + } + if err != nil || r.message.Seen || r.reads != 0 { + t.Fatalf("background update treated as view: seen=%v reads=%d err=%v", r.message.Seen, r.reads, err) + } + }) + } +} + +func TestSyncTruthPendingArrivalUpdatesRetryOnlyForOwnedUndeliveredMail(t *testing.T) { + user, mailbox, id := uuid.New(), uuid.New(), uuid.New() + for _, kind := range []string{"flags add", "flags remove", "folder", "full update"} { + t.Run(kind, func(t *testing.T) { + r := &syncTruthLookup{} + outbox := &syncTruthOutbox{user: user, mailbox: mailbox, id: id, pending: true} + s := &JobsService{UniboxRepository: r, ArrivalOutbox: outbox} + apply := func(owner, box uuid.UUID) error { + switch kind { + case "flags add": + return s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}}) + case "flags remove": + return s.HandleFlagsRemove(t.Context(), &models.JobEventFlags{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}}) + case "folder": + return s.HandleFolderUpdate(t.Context(), &models.JobEventFolderUpdate{UserID: owner, EmailID: box, ID: id, Folder: models.FolderArchive}) + default: + return s.HandleUpdateEmail(t.Context(), &models.JobEventEmailUpdate{UserID: owner, EmailID: box, ID: id, Flags: []string{models.FlagSeen}}) + } + } + if err := apply(user, mailbox); !errors.Is(err, ErrSyncArrivalPending) { + t.Fatalf("pending arrival was acknowledged: %v", err) + } + if err := apply(uuid.New(), mailbox); err != nil { + t.Fatalf("unowned update retried: %v", err) + } + if err := apply(user, uuid.New()); err != nil { + t.Fatalf("other-mailbox update retried: %v", err) + } + outbox.pending = false + if err := apply(user, mailbox); err != nil { + t.Fatalf("deliberately hidden/deleted mail retried: %v", err) + } + outbox.err = errors.New("marker unavailable") + if err := apply(user, mailbox); !errors.Is(err, outbox.err) { + t.Fatalf("marker failure swallowed: %v", err) + } + }) + } +} + +func TestSyncTruthArrivalFinishingDuringLookupStillGetsUpdate(t *testing.T) { + user, mailbox, id := uuid.New(), uuid.New(), uuid.New() + r := &syncTruthLookup{message: &models.EmailMessageStoreData{EmailID: mailbox}, appearOn: 3} + s := &JobsService{UniboxRepository: r, EmailRepository: warmupInboxEmailRepo{}, ArrivalOutbox: &syncTruthOutbox{user: user, mailbox: mailbox, id: id}} + if err := s.HandleFlagsAdd(t.Context(), &models.JobEventFlags{UserID: user, EmailID: mailbox, ID: id, Flags: []string{models.FlagSeen}}); err != nil { + t.Fatal(err) + } + if !r.message.Seen { + t.Fatal("arrival committed between reads lost the later Seen update") + } +} diff --git a/internal/app/consumer/unibox_seen_sync_test.go b/internal/app/consumer/unibox_seen_sync_test.go index 070ffe858..a9ebc56cf 100644 --- a/internal/app/consumer/unibox_seen_sync_test.go +++ b/internal/app/consumer/unibox_seen_sync_test.go @@ -21,6 +21,10 @@ func (r *seenSyncRepo) GetByID(context.Context, uuid.UUID, uuid.UUID) (*models.E return r.stored, nil } +func (r *seenSyncRepo) GetForSync(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (*models.EmailMessageStoreData, error) { + return r.stored, nil +} + func (r *seenSyncRepo) UpdateEntry(_ context.Context, _, _, _ uuid.UUID, e *repository.UpdateUniboxEntry) error { r.updates = append(r.updates, *e) return nil diff --git a/internal/app/worker/wmail/admit.go b/internal/app/worker/wmail/admit.go index d49c0a65a..e6dba5690 100644 --- a/internal/app/worker/wmail/admit.go +++ b/internal/app/worker/wmail/admit.go @@ -134,15 +134,21 @@ func (w *WMail) laneOf(ctx context.Context, key string, msg *models.EmailMessage // beginTick releases an expired hold so the state reports "within budget" // before the pass looks at anything. func (w *WMail) beginTick() { + w.tracker.tickComplete = false w.tracker.release(time.Now()) } -// endTick folds the pass into the relayed state. deferred is the number of -// live messages still waiting on the server; it is reported as-is (not -// accumulated) so a pass that admits everything zeroes it. -func (w *WMail) endTick(stats *tickStats) { - w.tracker.setDeferred(stats.deferred) - w.tracker.touch(time.Now()) +// endTick retains unmeasured backlog until a complete pass confirms catch-up. +func (w *WMail) endTick(ctx context.Context, stats *tickStats, complete bool) { + complete = complete && ctx.Err() == nil && !stats.aborted && stats.deferred == 0 && w.tracker.state.BackfillCursor.GoogleRecovery == nil + w.tracker.tickComplete = complete + if complete { + w.tracker.setDeferred(0) + w.tracker.touch(time.Now()) + return + } + w.tracker.setDeferred(max(w.tracker.state.Deferred, stats.deferred)) + w.tracker.flush(time.Now()) } // storeNew is the shared tail of every provider's new-message path: record diff --git a/internal/app/worker/wmail/start_sync.go b/internal/app/worker/wmail/start_sync.go index 3c7755021..8208e2bc9 100644 --- a/internal/app/worker/wmail/start_sync.go +++ b/internal/app/worker/wmail/start_sync.go @@ -78,11 +78,18 @@ func (w *WMail) nextSyncDelay(base time.Duration, last *errx.MailError) time.Dur // so one mailbox's bad server response must not take down every other // account's sync and send loops. func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) { + if w.tracker != nil { + w.tracker.tickComplete = false + } defer func() { if r := recover(); r != nil { + if w.tracker != nil { + w.tracker.tickComplete = false + } err := fmt.Errorf("mail sync panic: %v", r) w.CaptureError(err) log.Error().Err(err).Str("email_id", w.ID.String()).Msg("mail sync panicked") + result = &errx.MailError{Code: errx.MailErrorCodeImapUnknown, Type: errx.MailErrorWarning, Message: "mail sync pass interrupted"} } }() if err := w.SyncMail(ctx); err != nil { @@ -100,7 +107,9 @@ func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) { log.Warn().Err(err).Str("email_id", w.ID.String()).Msg("mail sync error") return err } - w.transportFailures = 0 + if w.tracker != nil && w.tracker.tickComplete { + w.transportFailures = 0 + } return nil } diff --git a/internal/app/worker/wmail/sync_google.go b/internal/app/worker/wmail/sync_google.go index 8b6b40fda..b81d626d4 100644 --- a/internal/app/worker/wmail/sync_google.go +++ b/internal/app/worker/wmail/sync_google.go @@ -29,7 +29,8 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError { stats := &tickStats{} w.googleTick = stats w.googleFolders = nil - defer w.endTick(stats) + complete := false + defer func() { w.endTick(ctx, stats, complete) }() if !w.retryUnmap(ctx) { return nil } @@ -40,7 +41,7 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError { return nil } - newHistoryID, err := w.GoogleData.Client.FetchHistory(ctx, w.GoogleData.LastHistoryID) + newHistoryID, caughtUp, err := w.GoogleData.Client.FetchHistoryPass(ctx, w.GoogleData.LastHistoryID) if errors.Is(err, goog.ErrHistoryExpired) { baseline, berr := w.GoogleData.Client.HistoryBaseline(ctx) if berr != nil { @@ -81,10 +82,11 @@ func (w *WMail) SyncGoogle(ctx context.Context) *errx.MailError { if !stats.aborted { // The pass's own work is done; only a refusal the owner has to act // on is worth failing it for. - if merr := w.googleReconcileFolders(ctx, time.Now(), stats); merr != nil && merr.Type == errx.MailErrorCritical { + if merr := w.googleReconcileFolders(ctx, time.Now(), stats); merr != nil { return merr } } + complete = caughtUp return nil } @@ -338,6 +340,7 @@ func (w *WMail) googleBackfill(ctx context.Context, stats *tickStats) *errx.Mail } ids, next, err := w.GoogleData.Client.ListMessages(ctx, q, st.BackfillCursor.PageToken, googleBackfillPage) if err != nil { + stats.aborted = true var errMail *errx.MailError if errors.As(err, &errMail) { return errMail @@ -364,6 +367,7 @@ func (w *WMail) googleBackfill(ctx context.Context, stats *tickStats) *errx.Mail } msg, err := w.GoogleData.Client.GetMessage(ctx, id) if err != nil { + stats.aborted = true var errMail *errx.MailError if errors.As(err, &errMail) { return errMail @@ -436,6 +440,7 @@ func (w *WMail) googleReconcileFolders(ctx context.Context, now time.Time, stats for page := 0; page < config.GmailFolderReconcilePages; page++ { ids, next, err := w.GoogleData.Client.ListLabelMessages(ctx, goog.Inbox, q, token, 500) if err != nil { + stats.aborted = true return w.googleReconcileError(err) } for _, id := range ids { @@ -479,6 +484,7 @@ func (w *WMail) googleReconcileFolders(ctx context.Context, now time.Time, stats labels, found, err := w.GoogleData.Client.MessageLabels(ctx, m.ProviderID) if err != nil { if gmailRetryable(err) { + stats.aborted = true return w.googleReconcileError(err) } // One message Gmail refuses must not hold up every row after it. diff --git a/internal/app/worker/wmail/sync_graph.go b/internal/app/worker/wmail/sync_graph.go index d34558db7..1efb5bf9a 100644 --- a/internal/app/worker/wmail/sync_graph.go +++ b/internal/app/worker/wmail/sync_graph.go @@ -25,11 +25,14 @@ func (w *WMail) SyncGraph(ctx context.Context) *errx.MailError { w.beginTick() stats := &tickStats{} w.graphTick = stats + complete := false + defer func() { w.endTick(ctx, stats, complete) }() if !w.retryUnmap(ctx) { return nil } - if err := w.GraphData.Client.Sync(ctx); err != nil { + caughtUp, err := w.GraphData.Client.SyncPass(ctx) + if err != nil { var mailErr *errx.MailError if errors.As(err, &mailErr) { return mailErr @@ -42,7 +45,7 @@ func (w *WMail) SyncGraph(ctx context.Context) *errx.MailError { return merr } } - w.endTick(stats) + complete = caughtUp return nil } @@ -186,6 +189,7 @@ func (w *WMail) graphBackfill(ctx context.Context, stats *tickStats) *errx.MailE return mailErr } w.CaptureError(err) + stats.aborted = true return nil } for _, full := range msgs { diff --git a/internal/app/worker/wmail/sync_imap.go b/internal/app/worker/wmail/sync_imap.go index c775013ff..bed4867f2 100644 --- a/internal/app/worker/wmail/sync_imap.go +++ b/internal/app/worker/wmail/sync_imap.go @@ -32,6 +32,8 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { } w.beginTick() stats := &tickStats{} + complete := false + defer func() { w.endTick(ctx, stats, complete) }() if !w.retryUnmap(ctx) { return nil } @@ -78,6 +80,7 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { // mod-sequences where the server has CONDSTORE, UIDNEXT where it does not // (Outlook.com, Microsoft 365 over IMAP, Yahoo, many hosted servers). condStore := client.HasCondStore() + caughtUp := true for i := range folders { box := &folders[i] @@ -136,6 +139,9 @@ func (w *WMail) Sync(ctx context.Context) *errx.MailError { // The pass was aborted before this folder; hold its cursor too. fullyProcessed = false } + if !fullyProcessed { + caughtUp = false + } if changed || !slices.Equal(befBox.Attrs, box.Attrs) { // The stored cursor only moves once every change up to it was @@ -238,7 +244,7 @@ outer: } } - w.endTick(stats) + complete = caughtUp return nil } diff --git a/internal/app/worker/wmail/sync_imap_flags.go b/internal/app/worker/wmail/sync_imap_flags.go index d870fbfa2..3145fcaba 100644 --- a/internal/app/worker/wmail/sync_imap_flags.go +++ b/internal/app/worker/wmail/sync_imap_flags.go @@ -73,7 +73,8 @@ func (w *WMail) imapScanFlags(ctx context.Context, box *models.Mailbox, stats *t return err } if stats.aborted { - break + scan.at = now + return nil } } } diff --git a/internal/app/worker/wmail/sync_state.go b/internal/app/worker/wmail/sync_state.go index 32c3d0693..d896d05df 100644 --- a/internal/app/worker/wmail/sync_state.go +++ b/internal/app/worker/wmail/sync_state.go @@ -16,10 +16,11 @@ const syncStateHeartbeat = 10 * time.Minute // of the tick; the consumer persists it and the loader hands it back on the // next assignment. type syncTracker struct { - state models.SyncState - dirty bool - lastSent time.Time - emit func(models.SyncState) error + state models.SyncState + dirty bool + lastSent time.Time + emit func(models.SyncState) error + tickComplete bool } func newSyncTracker(seed *models.SyncState, emit func(models.SyncState) error) *syncTracker { @@ -85,6 +86,11 @@ func (t *syncTracker) setFoldersSkipped(cap, conflict int) { func (t *syncTracker) touch(now time.Time) { stamp := now t.state.LastSyncedAt = &stamp + t.flush(now) +} + +// flush publishes safe progress without implying a successful provider check. +func (t *syncTracker) flush(now time.Time) { if !t.dirty && now.Sub(t.lastSent) < syncStateHeartbeat { return } diff --git a/internal/app/worker/wmail/sync_truth_test.go b/internal/app/worker/wmail/sync_truth_test.go new file mode 100644 index 000000000..9bc905bfc --- /dev/null +++ b/internal/app/worker/wmail/sync_truth_test.go @@ -0,0 +1,355 @@ +package wmail + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/client/goog" + "github.com/warmbly/warmbly/internal/client/msgraph" + "github.com/warmbly/warmbly/internal/client/smtpimap/imap" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +func seedSyncTruth(w *WMail, relayed *[]models.SyncState) time.Time { + previous := time.Now().Add(-time.Hour) + w.tracker = newSyncTracker(&models.SyncState{BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: &previous, Deferred: 5}, func(st models.SyncState) error { + *relayed = append(*relayed, st) + return nil + }) + return previous +} + +func assertNoSyncSuccess(t *testing.T, w *WMail, previous time.Time) { + t.Helper() + if w.tracker.state.LastSyncedAt == nil || !w.tracker.state.LastSyncedAt.Equal(previous) || w.tracker.state.Deferred != 5 { + t.Fatalf("incomplete pass replaced successful evidence/backlog: %+v", w.tracker.state) + } +} + +func assertSyncCatchUp(t *testing.T, w *WMail, previous time.Time) { + t.Helper() + if w.tracker.state.LastSyncedAt == nil || !w.tracker.state.LastSyncedAt.After(previous) || w.tracker.state.Deferred != 0 { + t.Fatalf("successful catch-up did not replace old evidence: %+v", w.tracker.state) + } +} + +func TestSyncTruthGoogleFailureHeartbeatAndRecovery(t *testing.T) { + failing := true + srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, _ *http.Request) { + out.Header().Set("Content-Type", "application/json") + if failing { + out.WriteHeader(http.StatusServiceUnavailable) + _, _ = out.Write([]byte(`{"error":{"code":503,"message":"retry later"}}`)) + return + } + _, _ = out.Write([]byte(`{"historyId":"200"}`)) + })) + defer srv.Close() + var events []captured + w := newGoogleTestMail(t, srv, &events) + w.GoogleData.LastHistoryID = 100 + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + for range 2 { + w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat) + if err := w.SyncGoogle(t.Context()); err == nil { + t.Fatal("expected provider error") + } + assertNoSyncSuccess(t, w, previous) + } + if len(relayed) != 2 { + t.Fatalf("failed passes must relay safe heartbeat state: %d", len(relayed)) + } + for _, st := range relayed { + if !st.LastSyncedAt.Equal(previous) || st.Deferred != 5 { + t.Fatalf("false success escaped via heartbeat: %+v", st) + } + } + failing = false + if err := w.SyncGoogle(t.Context()); err != nil { + t.Fatal(err) + } + assertSyncCatchUp(t, w, previous) +} + +type syncTruthMap struct { + repository.EmailMessageMapRepository + failGet bool + failDel bool +} + +func (r *syncTruthMap) Get(context.Context, uuid.UUID, uuid.UUID, string) (*repository.EmailMessageData, error) { + if r.failGet { + return nil, errors.New("map unavailable") + } + return nil, nil +} + +func (r *syncTruthMap) Del(context.Context, uuid.UUID, uuid.UUID, string, uuid.UUID) error { + if r.failDel { + return errors.New("unmap unavailable") + } + return nil +} + +func TestSyncTruthGraphAbortedAdmissionAndCheckpoint(t *testing.T) { + for _, failure := range []string{"map", "checkpoint"} { + t.Run(failure, func(t *testing.T) { + g := newFakeGraph() + var events []captured + var relayed []models.SyncState + w := newGraphTestMail(t, g.serve(t), &events, &relayed) + previous := seedSyncTruth(w, &relayed) + before := w.GraphData.Client.DeltaLinks[msgraph.FolderInbox] + r := &syncTruthMap{failGet: failure == "map"} + w.EmailMessageMapRepository = r + failing := true + w.onEvent = func(kind models.JobEventType, _ any) error { + if failing && failure == "checkpoint" && kind == models.JobEventTypeGraphDeltaUpdate { + return errors.New("checkpoint unavailable") + } + return nil + } + for range 2 { + if failure == "map" { + g.live = []string{"new-arrival"} + } + w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat) + _ = w.SyncGraph(t.Context()) + assertNoSyncSuccess(t, w, previous) + if w.GraphData.Client.DeltaLinks[msgraph.FolderInbox] != before { + t.Fatal("failed/aborted page advanced local checkpoint") + } + } + failing, r.failGet = false, false + if err := w.SyncGraph(t.Context()); err != nil { + t.Fatal(err) + } + assertSyncCatchUp(t, w, previous) + }) + } +} + +func TestSyncTruthRetryUnmapAndCancellation(t *testing.T) { + for _, provider := range []string{"google", "graph", "imap"} { + t.Run(provider, func(t *testing.T) { + w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{}) + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + w.unmapPending = map[string]uuid.UUID{"failed-arrival": uuid.New()} + w.EmailMessageMapRepository = &syncTruthMap{failDel: true} + switch provider { + case "google": + _ = w.SyncGoogle(t.Context()) + case "graph": + _ = w.SyncGraph(t.Context()) + case "imap": + _ = w.Sync(t.Context()) + } + assertNoSyncSuccess(t, w, previous) + if len(relayed) != 1 { + t.Fatal("retry-unmap blockage did not relay safe state") + } + }) + } + w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{}) + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _ = w.Sync(ctx) + assertNoSyncSuccess(t, w, previous) +} + +func TestSyncTruthImapFlagFailureRetainsBothUpdatesAndSuccessEvidence(t *testing.T) { + box := &models.Mailbox{Name: "INBOX", UIDValidity: 7, UIDNext: 101} + conn := &fakeImapConn{noCondStore: true, folders: []models.Mailbox{*box}, flags: map[uint32]imap.FlagState{ + 1: {MessageID: "one", Flags: nil}, 2: {MessageID: "two", Flags: nil}, + }} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, box) + w.EmailMessageMapRepository = knownMessageMap{id: uuid.NewString()} + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + conn.flags[1], conn.flags[2] = imap.FlagState{MessageID: "one", Flags: []string{models.FlagSeen}}, imap.FlagState{MessageID: "two", Flags: []string{models.FlagSeen}} + w.flagScan[box.Name].at = time.Now().Add(-2 * config.ImapFlagScanInterval) + failing := true + w.onEvent = func(kind models.JobEventType, body any) error { + if kind == models.JobEventTypeEmailUpdate && failing { + return errors.New("publisher unavailable") + } + *events = append(*events, captured{eventType: kind, body: body}) + return nil + } + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + assertNoSyncSuccess(t, w, previous) + for uid := range conn.flags { + if w.flagScan[box.Name].flags[uid] != flagFingerprint(nil) { + t.Fatal("failed/unvisited flags were absorbed into baseline") + } + } + failing = false + w.flagScan[box.Name].at = time.Now().Add(-2 * config.ImapFlagScanInterval) + if err := w.Sync(t.Context()); err != nil { + t.Fatal(err) + } + updates := 0 + for _, e := range *events { + if e.eventType == models.JobEventTypeEmailUpdate { + updates++ + } + } + if updates != 2 { + t.Fatalf("unchanged provider flags did not retry both updates: %d", updates) + } + assertSyncCatchUp(t, w, previous) +} + +func TestSyncTruthPanicDoesNotResetTransportFailure(t *testing.T) { + w, _ := newIMAPTestMail(&fakeImapConn{}, &fixedBudget{}, &models.Mailbox{}) + w.EmailType = models.InboxProviderGoogle + w.GoogleData = nil // Fault inside the provider pass, after beginTick. + w.transportFailures = 2 + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + if err := w.syncOnce(t.Context()); err == nil { + t.Fatal("recovered panic returned successful outcome") + } + assertNoSyncSuccess(t, w, previous) + if w.transportFailures != 2 { + t.Fatal("panic reset existing transport warning/backoff") + } +} + +func TestSyncTruthBoundedAndPinnedProviderPasses(t *testing.T) { + for _, provider := range []string{"google", "graph"} { + for _, outcome := range []string{"bounded", "pinned", "stop"} { + t.Run(provider+"/"+outcome, func(t *testing.T) { + caughtUp := false + srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, r *http.Request) { + out.Header().Set("Content-Type", "application/json") + if provider == "google" { + if caughtUp { + _, _ = out.Write([]byte(`{"historyId":"200"}`)) + return + } + if outcome == "bounded" { + _, _ = out.Write([]byte(`{"historyId":"200","nextPageToken":"next"}`)) + return + } + _, _ = out.Write([]byte(`{"historyId":"200","history":[{"id":"150","messagesAdded":[{"message":{"id":"new-arrival","threadId":"thread"}}]}]}`)) + return + } + folder := msgraph.FolderInbox + if caughtUp { + _, _ = fmt.Fprintf(out, `{"value":[],"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/%s/messages/delta?final=1"}`, folder) + return + } + if outcome == "bounded" { + _, _ = fmt.Fprintf(out, `{"value":[],"@odata.nextLink":"https://graph.microsoft.com%s?next=1"}`, r.URL.Path) + return + } + _, _ = fmt.Fprintf(out, `{"value":[{"id":"new-arrival","isRead":false}],"@odata.deltaLink":"https://graph.microsoft.com/v1.0/me/mailFolders/%s/messages/delta?final=1"}`, folder) + })) + defer srv.Close() + var events []captured + var relayed []models.SyncState + var w *WMail + var sync func(context.Context) *errx.MailError + if provider == "google" { + w = newGoogleTestMail(t, srv, &events) + w.GoogleData.LastHistoryID = 100 + w.GoogleData.Client.OnMessageAdded = func(context.Context, string, string) (bool, error) { + if outcome == "stop" { + return false, goog.ErrStop + } + return false, nil + } + sync = w.SyncGoogle + } else { + w = newGraphTestMail(t, srv, &events, &relayed) + w.GraphData.Client.OnMessageSeen = func(context.Context, string, string, bool) (bool, error) { + if outcome == "stop" { + return false, msgraph.ErrStop + } + return false, nil + } + sync = w.SyncGraph + } + previous := seedSyncTruth(w, &relayed) + w.transportFailures = 2 + if err := sync(t.Context()); err != nil { + t.Fatal(err) + } + assertNoSyncSuccess(t, w, previous) + caughtUp = true + if err := sync(t.Context()); err != nil { + t.Fatal(err) + } + assertSyncCatchUp(t, w, previous) + }) + } + } +} + +func TestSyncTruthGoogleCheckpointFailureAndDirtyRelayRetry(t *testing.T) { + failingProvider := false + srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, _ *http.Request) { + out.Header().Set("Content-Type", "application/json") + if failingProvider { + out.WriteHeader(503) + _, _ = out.Write([]byte(`{"error":{"code":503,"message":"injected"}}`)) + return + } + _, _ = out.Write([]byte(`{"historyId":"200"}`)) + })) + defer srv.Close() + var events []captured + w := newGoogleTestMail(t, srv, &events) + w.GoogleData.LastHistoryID = 100 + var relayed []models.SyncState + previous := seedSyncTruth(w, &relayed) + w.transportFailures = 2 + w.onEvent = func(kind models.JobEventType, _ any) error { + if kind == models.JobEventTypeHistoryIDUpdate { + return errors.New("checkpoint unavailable") + } + return nil + } + if err := w.syncOnce(t.Context()); err != nil { + t.Fatal(err) + } + assertNoSyncSuccess(t, w, previous) + if w.GoogleData.LastHistoryID != 100 || w.transportFailures != 2 { + t.Fatal("failed checkpoint was installed or claimed recovery") + } + failingProvider = true + expired := time.Now().Add(-time.Minute) + w.tracker.state.ThrottledUntil = &expired + w.tracker.lastSent = time.Now().Add(-2 * syncStateHeartbeat) + w.tracker.emit = func(st models.SyncState) error { return errors.New("state relay unavailable") } + _ = w.SyncGoogle(t.Context()) + assertNoSyncSuccess(t, w, previous) + if !w.tracker.dirty || w.tracker.state.ThrottledUntil != nil { + t.Fatal("failed relay forgot dirty throttle progress") + } + w.tracker.emit = func(st models.SyncState) error { relayed = append(relayed, st); return nil } + _ = w.SyncGoogle(t.Context()) + assertNoSyncSuccess(t, w, previous) + if w.tracker.dirty || relayed[len(relayed)-1].ThrottledUntil != nil { + t.Fatal("safe dirty-state relay did not retry") + } +} diff --git a/internal/client/goog/history.go b/internal/client/goog/history.go index 7865baabc..61f9834ae 100644 --- a/internal/client/goog/history.go +++ b/internal/client/goog/history.go @@ -14,6 +14,8 @@ import ( // returns the checkpoint reached so far and a nil error. var ErrStop = errors.New("goog: stop") +var errHistoryIncomplete = errors.New("goog: history pass incomplete") + var ErrHistoryExpired = errors.New("goog: history expired; resynchronization required") var ErrRecoveryPageExpired = errors.New("goog: recovery page token expired") @@ -35,6 +37,20 @@ const historyPagesPerPass = 10 // continues, so later records that are admitted (a reply, say) still land and // the deferred ones are re-offered next tick. func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64, error) { + checkpoint, _, err := c.FetchHistoryPass(ctx, lastHistoryID) + return checkpoint, err +} + +// FetchHistoryPass distinguishes a bounded or pinned walk from full live catch-up. +func (c *Client) FetchHistoryPass(ctx context.Context, lastHistoryID uint64) (uint64, bool, error) { + checkpoint, err := c.fetchHistory(ctx, lastHistoryID) + if errors.Is(err, errHistoryIncomplete) { + return checkpoint, false, nil + } + return checkpoint, err == nil, err +} + +func (c *Client) fetchHistory(ctx context.Context, lastHistoryID uint64) (uint64, error) { if lastHistoryID == 0 { return c.HistoryBaseline(ctx) } @@ -62,7 +78,7 @@ func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64 } added, err := c.OnMessageAdded(ctx, m.Message.Id, m.Message.ThreadId) if errors.Is(err, ErrStop) { - return checkpoint, nil + return checkpoint, errHistoryIncomplete } if err != nil { return checkpoint, err @@ -96,13 +112,14 @@ func (c *Client) FetchHistory(ctx context.Context, lastHistoryID uint64) (uint64 if resp.NextPageToken == "" { if !pinned { checkpoint = resp.HistoryId + return checkpoint, nil } - break + return checkpoint, errHistoryIncomplete } call.PageToken(resp.NextPageToken) } - return checkpoint, nil + return checkpoint, errHistoryIncomplete } func (c *Client) HistoryBaseline(ctx context.Context) (uint64, error) { diff --git a/internal/client/msgraph/delta.go b/internal/client/msgraph/delta.go index df1baf3ba..63f12bf78 100644 --- a/internal/client/msgraph/delta.go +++ b/internal/client/msgraph/delta.go @@ -19,6 +19,8 @@ func itoa(n int) string { return strconv.Itoa(n) } // where it was last persisted. var ErrStop = errors.New("msgraph: stop") +var errDeltaIncomplete = errors.New("msgraph: delta pass incomplete") + // deltaSelect keeps delta pages light: we only need the id, read state, and the // @removed marker to decide add/update vs remove. Full envelope + body + headers // are hydrated per admitted message via FetchMessage. @@ -64,15 +66,26 @@ var BackfillFolders = []string{FolderInbox, FolderSent, FolderArchive, FolderDra // fits the disposable worker natively: no webhook endpoint, no subscription // lifecycle, just a cursor the control plane persists via OnDelta. func (c *Client) Sync(ctx context.Context) error { + _, err := c.SyncPass(ctx) + return err +} + +// SyncPass reports catch-up only when every tracked folder reaches an unpinned delta link. +func (c *Client) SyncPass(ctx context.Context) (bool, error) { + complete := true for _, folder := range TrackedFolders { if err := c.syncFolder(ctx, folder); err != nil { if errors.Is(err, ErrStop) { - return nil + return false, nil } - return err + if errors.Is(err, errDeltaIncomplete) { + complete = false + continue + } + return false, err } } - return nil + return complete, nil } func (c *Client) syncFolder(ctx context.Context, folder string) error { @@ -97,7 +110,7 @@ func (c *Client) syncFolder(ctx context.Context, folder string) error { case pg.DeltaLink != "": return c.saveCursor(ctx, folder, pg.DeltaLink) default: - return nil + return errDeltaIncomplete } } } @@ -139,19 +152,21 @@ func (c *Client) syncFolder(ctx context.Context, folder string) error { if !pinned { return c.saveCursor(ctx, folder, pg.DeltaLink) } - return nil + return errDeltaIncomplete default: - return nil + return errDeltaIncomplete } } - return nil + return errDeltaIncomplete } func (c *Client) saveCursor(ctx context.Context, folder, link string) error { - c.DeltaLinks[folder] = link if c.OnDelta != nil { - return c.OnDelta(ctx, folder, link) + if err := c.OnDelta(ctx, folder, link); err != nil { + return err + } } + c.DeltaLinks[folder] = link return nil } diff --git a/internal/repository/pg_unibox.go b/internal/repository/pg_unibox.go index 724c92636..55f269aba 100644 --- a/internal/repository/pg_unibox.go +++ b/internal/repository/pg_unibox.go @@ -39,6 +39,7 @@ type UniboxRepository interface { UpdateEntry(ctx context.Context, userID, emailID, id uuid.UUID, e *UpdateUniboxEntry) error GetIncoming(ctx context.Context, userID uuid.UUID, limit int, cursor string) (*models.MailSearchResult, error) GetByID(ctx context.Context, userID, id uuid.UUID) (*models.EmailMessageStoreData, error) + GetForSync(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error) // GetByIDForOrg is the org-scoped read for the unibox detail view: any // member with unibox access can open a message in the org-wide list. It // returns the row plus the mailbox OWNER's user_id, which the S3 body key @@ -455,14 +456,35 @@ func (r *uniboxRepository) GetIncoming(ctx context.Context, userID uuid.UUID, li var ErrEmailNotFound = errors.New("email not found") func (r *uniboxRepository) GetByID(ctx context.Context, userID, id uuid.UUID) (*models.EmailMessageStoreData, error) { + e, err := r.getMessage(ctx, userID, uuid.Nil, id) + if err != nil { + return nil, err + } + if !e.Seen { + _ = r.MarkSeen(ctx, userID, id, true) + e.Seen = true + } + return e, nil +} + +// GetForSync reads provider state without treating background work as a user view. +func (r *uniboxRepository) GetForSync(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error) { + if emailID == uuid.Nil { + return nil, ErrEmailNotFound + } + return r.getMessage(ctx, userID, emailID, id) +} + +func (r *uniboxRepository) getMessage(ctx context.Context, userID, emailID, id uuid.UUID) (*models.EmailMessageStoreData, error) { query := fmt.Sprintf(` SELECT %s FROM unibox_emails WHERE user_id = $1 AND id = $2 + AND ($3::uuid = '00000000-0000-0000-0000-000000000000' OR email_id = $3) `, strings.Join(mailFieldsFull, ", ")) var e models.EmailMessageStoreData - err := r.db.QueryRow(ctx, query, userID, id).Scan( + err := r.db.QueryRow(ctx, query, userID, id, emailID).Scan( &e.ID, &e.EmailID, &e.Mailbox, &e.FolderPath, &e.ThreadID, &e.MessageID, &e.GmailID, &e.ParentID, &e.UID, &e.ModSeq, &e.Flags, &e.BCC, &e.CC, &e.FromAddr, &e.InReplyTo, &e.ReplyTo, @@ -476,12 +498,6 @@ func (r *uniboxRepository) GetByID(ctx context.Context, userID, id uuid.UUID) (* return nil, err } - // Auto-mark as seen - if !e.Seen { - _ = r.MarkSeen(ctx, userID, id, true) - e.Seen = true - } - return &e, nil } diff --git a/web/src/components/app/emails/SyncStatusCard.test.tsx b/web/src/components/app/emails/SyncStatusCard.test.tsx new file mode 100644 index 000000000..79f3a68b3 --- /dev/null +++ b/web/src/components/app/emails/SyncStatusCard.test.tsx @@ -0,0 +1,70 @@ +import { act, cleanup, render, screen } from "@testing-library/react"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type EmailSync from "@/lib/api/models/app/emails/SyncState"; +import SyncStatusCard from "./SyncStatusCard"; + +const fixtures = vi.hoisted(() => ({ data: undefined as EmailSync | undefined })); +vi.mock("@/lib/api/hooks/app/emails/useSync", () => ({ default: () => ({ data: fixtures.data, isPending: false }) })); +vi.mock("@/lib/api/hooks/app/emails/useUpdateSyncSkipFolders", () => ({ default: () => ({}) })); +vi.mock("@/hooks/context/confirm", () => ({ useConfirm: () => ({}) })); + +describe("truthful mailbox sync status", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-09T12:00:00Z")); + fixtures.data = { + state: { backfill_status: "complete", backfill_synced: 42, deferred: 0 }, + policy: { backfill_days: 30, backfill_messages: 2000, daily_messages: 500, org_daily_messages: 2000 }, + folders: [], skip_folders: [], + }; + }); + afterEach(() => { cleanup(); vi.useRealTimers(); }); + const draw = () => render(); + + it("does not treat a completed historical import as proof of live health", () => { + draw(); + expect(screen.getByText("Recent-mail import complete")).toBeInTheDocument(); + expect(screen.getByText("No successful sync check reported yet.")).toBeInTheDocument(); + expect(screen.getByText(/Import completion does not confirm live sync is current/)).toBeInTheDocument(); + expect(screen.queryByText(/Up to date/i)).not.toBeInTheDocument(); + }); + + it("ages a recent successful check into stale evidence without a new server payload", () => { + fixtures.data!.state!.last_synced_at = new Date("2026-10-09T11:59:00Z"); + draw(); + expect(screen.getByText("Last successful sync 1 min ago.")).toBeInTheDocument(); + act(() => { vi.advanceTimersByTime(21 * 60_000); }); + expect(screen.getByText("No recent successful sync check")).toBeInTheDocument(); + expect(screen.getByText(/Recovery is not yet confirmed by a recent successful check/)).toBeInTheDocument(); + expect(screen.queryByText(/Up to date/i)).not.toBeInTheDocument(); + }); + + it("shows ongoing recovery from the existing cursor instead of historical completion", () => { + fixtures.data!.state!.backfill_cursor = { google_recovery: { history_id: 100, since: new Date(), messages_done: false } }; + draw(); + expect(screen.getByText("Recovering mailbox sync")).toBeInTheDocument(); + expect(screen.queryByText("Recent-mail import complete")).not.toBeInTheDocument(); + }); + + it("keeps deferred evidence visible after the budget hold expires", () => { + fixtures.data!.state!.deferred = 5; + fixtures.data!.state!.throttled_until = new Date("2026-10-09T12:01:00Z"); + const view = draw(); + expect(screen.getByText(/Waiting on the sync budget until/)).toBeInTheDocument(); + act(() => { vi.advanceTimersByTime(2 * 60_000); }); + expect(screen.getByText("New mail was waiting on sync at the last check")).toBeInTheDocument(); + expect(screen.getByText(/Last observed backlog: 5 messages/)).toBeInTheDocument(); + fixtures.data!.state!.deferred = 0; + fixtures.data!.state!.last_synced_at = new Date(); + view.rerender(); + expect(screen.getByText("Last successful sync just now.")).toBeInTheDocument(); + expect(screen.queryByText(/Last observed backlog/)).not.toBeInTheDocument(); + }); + + it("describes import progress separately from the last successful live check", () => { + fixtures.data!.state!.backfill_status = "running"; + draw(); + expect(screen.getByText("Importing recent mail: 42 messages so far")).toBeInTheDocument(); + expect(screen.getByText("No successful sync check reported yet.")).toBeInTheDocument(); + }); +}); diff --git a/web/src/components/app/emails/SyncStatusCard.tsx b/web/src/components/app/emails/SyncStatusCard.tsx index 2a925bd17..2fb55bce0 100644 --- a/web/src/components/app/emails/SyncStatusCard.tsx +++ b/web/src/components/app/emails/SyncStatusCard.tsx @@ -1,4 +1,4 @@ -import { useState } from "react"; +import { useEffect, useState } from "react"; import { motion } from "framer-motion"; import { CheckCircle2Icon, DownloadIcon, HourglassIcon, PlusIcon, RefreshCwIcon } from "lucide-react"; import toast from "react-hot-toast/headless"; @@ -25,8 +25,8 @@ const REASON_COPY: Record = { priority_daily: "the daily limit for replies was reached", }; -function relative(iso: string | Date): string { - const diff = Date.now() - new Date(iso).getTime(); +function relative(iso: string | Date, now: number): string { + const diff = now - new Date(iso).getTime(); const m = Math.round(diff / 60_000); if (m < 1) return "just now"; if (m < 60) return `${m} min ago`; @@ -148,6 +148,11 @@ function SkipFoldersSection({ mailboxId, listed, skipped }: { mailboxId: string; } export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: string; provider?: string }) { + const [now, setNow] = useState(Date.now); + useEffect(() => { + const timer = setInterval(() => setNow(Date.now()), 60_000); + return () => clearInterval(timer); + }, []); const sync = useSync(mailboxId); const state = sync.data?.state ?? null; const policy = sync.data?.policy; @@ -162,15 +167,19 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str } if (!sync.data) return null; - const throttled = !!state?.throttled_until && new Date(state.throttled_until).getTime() > Date.now(); + const throttled = !!state?.throttled_until && new Date(state.throttled_until).getTime() > now; + const lastSuccess = state?.last_synced_at && new Date(state.last_synced_at).getTime(); + // The worker relays successful checks at least every ten minutes. + const stale = !!lastSuccess && now - lastSuccess > 20 * 60_000; + const recovering = !!state?.backfill_cursor?.google_recovery; const status = state?.backfill_status ?? "pending"; const cap = policy?.backfill_messages ?? 0; const synced = state?.backfill_synced ?? 0; const pct = cap > 0 ? Math.min(100, Math.round((synced / cap) * 100)) : 0; let headline: React.ReactNode; - let Icon = CheckCircle2Icon; - let tone = "text-emerald-600"; + let Icon = RefreshCwIcon; + let tone = "text-slate-500"; if (throttled && state?.throttled_until) { Icon = HourglassIcon; tone = "text-amber-600"; @@ -181,8 +190,19 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str {reason ? ({reason}) : null} ); + } else if (recovering) { + tone = "text-sky-600"; + headline = "Recovering mailbox sync"; + } else if ((state?.deferred ?? 0) > 0) { + Icon = HourglassIcon; + tone = "text-amber-600"; + headline = "New mail was waiting on sync at the last check"; + } else if (stale) { + tone = "text-amber-600"; + headline = "No recent successful sync check"; } else if (status === "complete") { - headline = state?.last_synced_at ? `Up to date, last checked ${relative(state.last_synced_at)}` : "Up to date"; + Icon = CheckCircle2Icon; + headline = "Recent-mail import complete"; } else if (status === "running") { Icon = DownloadIcon; tone = "text-sky-600"; @@ -214,16 +234,23 @@ export default function SyncStatusCard({ mailboxId, provider }: { mailboxId: str

{status === "complete" && policy - ? `Imported ${synced.toLocaleString()} message${synced === 1 ? "" : "s"} from the last ${policy.backfill_days} days. New mail syncs as it arrives.` + ? `Imported ${synced.toLocaleString()} message${synced === 1 ? "" : "s"} from the configured ${policy.backfill_days}-day window. Import completion does not confirm live sync is current.` : policy ? `The last ${policy.backfill_days} days come in newest first, up to ${cap.toLocaleString()} messages. New mail syncs alongside.` : null} - {throttled ? " Replies to your outreach keep syncing; the rest resumes automatically." : null} + {throttled ? " Mail is retried automatically within its sync budget." : null} +

+ +

+ {state?.last_synced_at + ? `Last successful sync ${relative(state.last_synced_at, now)}.` + : "No successful sync check reported yet."} + {stale ? " Recovery is not yet confirmed by a recent successful check." : null}

{(state?.deferred ?? 0) > 0 && (

- {state!.deferred.toLocaleString()} message{state!.deferred === 1 ? "" : "s"} waiting on the server. + Last observed backlog: {state!.deferred.toLocaleString()} message{state!.deferred === 1 ? "" : "s"} waiting on the server. The current total may have changed.

)} diff --git a/web/src/lib/api/hooks/app/emails/useSync.ts b/web/src/lib/api/hooks/app/emails/useSync.ts index ece52e272..acc031410 100644 --- a/web/src/lib/api/hooks/app/emails/useSync.ts +++ b/web/src/lib/api/hooks/app/emails/useSync.ts @@ -12,8 +12,8 @@ export default function useSync(id: string, enabled = true) { enabled: !!id && enabled, staleTime: 15_000, refetchInterval: (query) => { - const status = query.state.data?.state?.backfill_status; - return status === "running" || status === "pending" ? 20_000 : false; + const state = query.state.data?.state; + return state?.backfill_status === "running" || state?.backfill_status === "pending" || state?.backfill_cursor?.google_recovery ? 20_000 : 60_000; }, }); } diff --git a/web/src/lib/api/models/app/emails/SyncState.ts b/web/src/lib/api/models/app/emails/SyncState.ts index 9c7e76b70..7078a989d 100644 --- a/web/src/lib/api/models/app/emails/SyncState.ts +++ b/web/src/lib/api/models/app/emails/SyncState.ts @@ -12,6 +12,17 @@ export type SyncThrottleReason = export interface SyncState { backfill_status: SyncBackfillStatus; + backfill_cursor?: { + page_token?: string; + folders?: Record; + google_recovery?: { + history_id: number; + since: Date | string; + page_token?: string; + messages_done: boolean; + stored_after?: string; + }; + }; backfill_synced: number; backfill_since?: Date; backfill_started_at?: Date;