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>

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")
+	}
+}

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")
+	}
+}

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")
+	}
+}

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)