From 2f842cf96cca7459bd75c31f01beb7b5d2cf71e4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sun, 11 Oct 2026 05:25:49 +0000 Subject: [PATCH] feat: keep retired and cancelled sync attempts out of provider-failure evidence while preserving native recovery retries Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../worker/mailmanager/recovery_probe_test.go | 84 ++++++++++++ internal/app/worker/recovery_probe_test.go | 32 +++++ .../app/worker/wmail/recovery_probe_test.go | 129 ++++++++++++++++++ internal/app/worker/wmail/start_sync.go | 6 + 4 files changed, 251 insertions(+) create mode 100644 internal/app/worker/mailmanager/recovery_probe_test.go create mode 100644 internal/app/worker/recovery_probe_test.go create mode 100644 internal/app/worker/wmail/recovery_probe_test.go diff --git a/internal/app/worker/mailmanager/recovery_probe_test.go b/internal/app/worker/mailmanager/recovery_probe_test.go new file mode 100644 index 000000000..b7066b8ee --- /dev/null +++ b/internal/app/worker/mailmanager/recovery_probe_test.go @@ -0,0 +1,84 @@ +package mailmanager + +import ( + "net" + "testing" + "time" + + goimap "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapserver" + "github.com/emersion/go-imap/v2/imapserver/imapmemserver" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func TestRecoveryProbeFailedReplacementRetainsStoppedButReofferRecovers(t *testing.T) { + t.Setenv("DEPLOYMENT_MODE", "self_hosted") + events := 0 + m := NewMailManager(func(models.JobEventType, string, any) error { events++; return nil }, nil, nil, nil, nil, nil, nil) + d := smtpPayload() + stamp := time.Now().Add(-time.Hour) + d.Sync.State.LastSyncedAt = &stamp + if err := m.AddWMail(t.Context(), d); err != nil { + t.Fatal(err) + } + old := m.Get(d.ID) + t.Cleanup(func() { m.Terminate(d.ID) }) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr, port := ln.Addr().String(), ln.Addr().(*net.TCPAddr).Port + if err := ln.Close(); err != nil { + t.Fatal(err) + } + retry := *d + s := *d.SmtpImap + creds := *s.Credentials + creds.IMAP = &models.Service{Host: "127.0.0.1", Port: port, Username: "local", Password: "local", Security: models.MailSecurityNone} + s.Credentials = &creds + retry.SmtpImap, retry.ImapSync = &s, true + if err := m.AddWMail(t.Context(), &retry); err != errx.ErrMailServerUnreachable { + t.Fatalf("initial unavailable provider: %v", err) + } + if m.Get(d.ID) != old || !m.Has(d.ID) || old.Ctx.Err() == nil { + t.Fatal("failed replacement did not retain stopped pointer") + } + if _, ok := old.BeginExecution(); ok { + t.Fatal("stopped pointer admitted work") + } + if old.SameExecution(d) || old.SameExecution(&retry) { + t.Fatal("stopped pointer bypasses reload on republish") + } + old.CaptureError(errx.ErrMailServerUnreachable) + if events != 0 { + t.Fatal("retired pointer emitted a fresh transport report") + } + mem := imapmemserver.New() + u := imapmemserver.NewUser("local", "local") + if err := u.Create("INBOX", nil); err != nil { + t.Fatal(err) + } + mem.AddUser(u) + ln, err = net.Listen("tcp", addr) + if err != nil { + t.Fatal(err) + } + srv := imapserver.New(&imapserver.Options{Caps: goimap.CapSet{goimap.CapIMAP4rev1: {}}, InsecureAuth: true, NewSession: func(*imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { + return mem.NewSession(), nil, nil + }}) + go func() { _ = srv.Serve(ln) }() + defer srv.Close() + if err := m.AddWMail(t.Context(), &retry); err != nil { + t.Fatal(err) + } + next := m.Get(d.ID) + if next == nil || next == old || next.Ctx.Err() != nil || !next.SameExecution(&retry) { + t.Fatal("same reoffered credentials did not recover") + } + resumed := retry + next.ResumeInto(&resumed) + if resumed.Sync.State.LastSyncedAt == nil || !resumed.Sync.State.LastSyncedAt.Equal(stamp) || resumed.Sync.State.BackfillSynced != 7 { + t.Fatal("retry lost previous sync evidence") + } +} diff --git a/internal/app/worker/recovery_probe_test.go b/internal/app/worker/recovery_probe_test.go new file mode 100644 index 000000000..5aa493d3b --- /dev/null +++ b/internal/app/worker/recovery_probe_test.go @@ -0,0 +1,32 @@ +package worker + +import ( + "net" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func TestRecoveryProbeInitialDialFailureHasReportButNoState(t *testing.T) { + t.Setenv("DEPLOYMENT_MODE", "self_hosted") + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := ln.Addr().(*net.TCPAddr).Port + if err := ln.Close(); err != nil { + t.Fatal(err) + } + c := &capturedEvents{} + w := newLoadingWorker(c) + e := &models.AddWorkerEmail{ID: uuid.New(), UserID: uuid.New(), Type: models.InboxProviderSMTPIMAP, ImapSync: true, SmtpImap: &models.AddWorkerEmailSmtpImapData{Credentials: &models.SmtpImap{IMAP: &models.Service{Host: "127.0.0.1", Port: port, Security: models.MailSecurityNone}}}} + w.loadMailbox(t.Context(), e, &mailboxLoad{}) + if w.mailManager.Get(e.ID) != nil { + t.Fatal("failed initial dial installed mailbox") + } + if len(c.events) != 1 || c.events[0] != models.JobEventTypeEmailServerError || c.bodies[0].(models.EmailErrorEvent).ErrorCode != string(errx.MailErrorCodeServerUnreachable) { + t.Fatal("initial dial failure did not produce only a transport report") + } +} diff --git a/internal/app/worker/wmail/recovery_probe_test.go b/internal/app/worker/wmail/recovery_probe_test.go new file mode 100644 index 000000000..e965b9773 --- /dev/null +++ b/internal/app/worker/wmail/recovery_probe_test.go @@ -0,0 +1,129 @@ +package wmail + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/nodeevidence" +) + +func TestRecoveryProbeRetiredSyncIsNotProviderUnreachable(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + w := &WMail{Ctx: ctx, Cancel: cancel} + w.Stop() + observations := 0 + nodeevidence.Install(func(e nodeevidence.Event) { + if e.Name == nodeevidence.ProviderUnreachable { + observations++ + } + }) + defer nodeevidence.Install(nil) + w.StartSyncWorker(w.Ctx) + if observations != 0 { + t.Fatalf("retired generation emitted %d provider-unreachable observations without a provider call", observations) + } +} + +func TestRecoveryProbeCallerCancellationPreservesErrorWithoutProviderEvidence(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + reports, observations := 0, 0 + w := &WMail{Ctx: context.Background(), transportFailures: 2, onEvent: func(models.JobEventType, any) error { reports++; return nil }} + w.tracker = newSyncTracker(nil, func(models.SyncState) error { t.Fatal("cancelled pass published state"); return nil }) + nodeevidence.Install(func(nodeevidence.Event) { observations++ }) + defer nodeevidence.Install(nil) + if err := w.syncOnce(ctx); err != errx.ErrMailServerUnreachable { + t.Fatal("cancellation return identity changed") + } + if reports != 0 || observations != 0 || w.transportFailures != 2 || w.tracker.tickComplete || w.tracker.state.LastSyncedAt != nil { + t.Fatal("cancellation claimed provider failure or success") + } +} + +func TestRecoveryProbeMissingStateTransportPassThenRecovery(t *testing.T) { + conn := &fakeImapConn{folderErr: errx.ErrMailServerUnreachable} + w, events := newIMAPTestMail(conn, &fixedBudget{allow: 100}, &models.Mailbox{Name: "INBOX"}) + w.EmailType = models.InboxProviderSMTPIMAP + var relayed []models.SyncState + w.tracker = newSyncTracker(nil, func(st models.SyncState) error { relayed = append(relayed, st); return nil }) + for range 2 { + if err := w.syncOnce(t.Context()); err != errx.ErrMailServerUnreachable { + t.Fatal("transport error identity changed") + } + if w.tracker.state.LastSyncedAt != nil || w.tracker.tickComplete { + t.Fatal("failed pass invented provider completion") + } + } + if len(relayed) == 0 || relayed[0].LastSyncedAt != nil || len(mailboxEvents(*events, models.JobEventTypeEmailServerError)) != 2 { + t.Fatal("live failed pass did not flush safe state and error observations") + } + conn.folderErr = nil + if err := w.syncOnce(t.Context()); err != nil { + t.Fatal(err) + } + if w.tracker.state.LastSyncedAt == nil || !w.tracker.tickComplete || w.transportFailures != 0 { + t.Fatal("restored provider did not recover") + } +} + +func TestRecoveryProbeGraphMalformedPassRetainsCompletionAndRetries(t *testing.T) { + var broken atomic.Bool + broken.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, req *http.Request) { + out.Header().Set("Content-Type", "application/json") + if broken.Load() { + _, _ = out.Write([]byte(`{"value":`)) + return + } + _, _ = fmt.Fprintf(out, `{"value":[],"@odata.deltaLink":"https://graph.microsoft.com%s?$deltatoken=recovered"}`, req.URL.Path) + })) + defer srv.Close() + var events []captured + var relayed []models.SyncState + w := newGraphTestMail(t, srv, &events, &relayed) + previous := seedSyncTruth(w, &relayed) + w.transportFailures = 2 + if err := w.syncOnce(t.Context()); err != nil { + t.Fatal("non-MailError behavior changed") + } + assertNoSyncSuccess(t, w, previous) + if len(relayed) == 0 || w.transportFailures != 2 { + t.Fatal("incomplete pass failed safe heartbeat") + } + if len(mailboxEvents(events, models.JobEventTypeEmailServerError)) != 0 { + t.Fatal("raw Graph decode failure fabricated transport report") + } + broken.Store(false) + if err := w.syncOnce(t.Context()); err != nil { + t.Fatal(err) + } + assertSyncCatchUp(t, w, previous) +} + +func TestRecoveryProbeWrappedTransportRelayAndRetryFloor(t *testing.T) { + mailErr := *errx.ErrMailServerUnreachable + mailErr.RetryAfter = 12 * time.Minute + err := fmt.Errorf("bounded fake wrapper: %w", &mailErr) + var events []captured + w := &WMail{onEvent: func(kind models.JobEventType, body any) error { + events = append(events, captured{eventType: kind, body: body}) + return nil + }} + if reportable(err) { + t.Fatal("wrapped retryable condition escalated to application exception") + } + w.CaptureError(err) + if len(events) != 1 || events[0].eventType != models.JobEventTypeEmailServerError || events[0].body.(models.EmailErrorEvent).ErrorCode != string(errx.MailErrorCodeServerUnreachable) { + t.Fatal("wrapped MailError was not relayed") + } + if delay := w.nextSyncDelay(time.Minute, &mailErr); delay < mailErr.RetryAfter { + t.Fatal("provider retry interval shortened") + } +} diff --git a/internal/app/worker/wmail/start_sync.go b/internal/app/worker/wmail/start_sync.go index 394aedc5a..0bc00e8ef 100644 --- a/internal/app/worker/wmail/start_sync.go +++ b/internal/app/worker/wmail/start_sync.go @@ -93,6 +93,12 @@ func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) { } }() if err := w.SyncMail(ctx); err != nil { + w.lifecycleMu.Lock() + stopped := w.stopped + w.lifecycleMu.Unlock() + if stopped || ctx.Err() != nil { + return err + } switch err.Code { case errx.MailErrorCodeServerUnreachable, errx.MailErrorCodeConnectionLost: nodeevidence.Emit(nodeevidence.ProviderUnreachable, w.ID, 0, 1)