feat: avoid repeated failure scans and stop confirmed warmup lookup at the newest provider receipt

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

internal/repository/warmup_send_failure.go

@@ -26,10 +26,21 @@ const warmupSendFailuresSQL = `
 		ORDER BY t.updated_at DESC, t.id DESC LIMIT 1
 	) latest ON true
 	LEFT JOIN LATERAL (
-		SELECT MAX(c.completed_at) AS at
-		FROM tasks c JOIN warmup_tokens wt ON wt.task_id = c.id AND wt.sent_message_id <> ''
-		WHERE c.email_account_id = e.id AND c.task_type = 'warmup' AND c.status = 'completed' AND c.completed_at <= $3
+		SELECT c.completed_at AS at
+		FROM (
+			SELECT id, completed_at FROM tasks
+			WHERE email_account_id = e.id AND task_type = 'warmup' AND status = 'completed' AND completed_at <= $3
+			ORDER BY completed_at DESC
+		) c JOIN warmup_tokens wt ON wt.task_id = c.id AND wt.sent_message_id <> ''
+		ORDER BY c.completed_at DESC LIMIT 1
 	) confirmed ON true
+	LEFT JOIN LATERAL (
+		SELECT p.updated_at AS at
+		FROM tasks p JOIN task_failures pf ON pf.task_id = p.id
+		WHERE p.email_account_id = e.id AND p.task_type = 'warmup' AND p.status = 'failed'
+		  AND pf.message NOT LIKE $2 AND p.updated_at <= $3
+		ORDER BY p.updated_at DESC LIMIT 1
+	) non_loading ON latest.message LIKE $2
 	LEFT JOIN LATERAL (
 		WITH loading AS (
 			SELECT t.updated_at AS at,
@@ -39,11 +50,7 @@ const warmupSendFailuresSQL = `
 			  AND tf.message LIKE $2
 			  AND t.updated_at <= $3
 			  AND (confirmed.at IS NULL OR t.updated_at > confirmed.at)
-			  AND NOT EXISTS (
-				SELECT 1 FROM tasks p JOIN task_failures pf ON pf.task_id=p.id
-				WHERE p.email_account_id=e.id AND p.task_type='warmup' AND p.status='failed'
-				  AND pf.message NOT LIKE $2 AND p.updated_at >= t.updated_at AND p.updated_at <= $3
-			  )
+			  AND (non_loading.at IS NULL OR t.updated_at > non_loading.at)
 		)
 		SELECT MIN(at) AS at FROM loading
 		WHERE at >= COALESCE((SELECT MAX(at) FROM loading

internal/repository/warmup_send_failure_live_test.go

@@ -211,6 +211,41 @@ func TestLivePersistentMailboxLoadingFailure(t *testing.T) {
 	}
 }
 
+func TestLiveLoadingIncidentNonLoadingBoundary(t *testing.T) {
+	_, pool := liveContactDB(t)
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	for _, tc := range []struct {
+		name     string
+		boundary time.Duration
+		first    time.Duration
+	}{
+		{"before incident", -3 * time.Hour, -2 * time.Hour},
+		{"same timestamp", -80 * time.Minute, -40 * time.Minute},
+		{"between failures", -79 * time.Minute, -40 * time.Minute},
+		{"future failure", 10 * time.Minute, -2 * time.Hour},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newWarmupUsageFixture(t, pool)
+			failed := func(offset time.Duration, message string) {
+				id := uuid.New()
+				f.exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,updated_at) VALUES($1,'warmup',$2,'failed','',$3)`, id, f.account, now.Add(offset))
+				f.exec(`INSERT INTO task_failures(task_id,title,message) VALUES($1,'private',$2)`, id, message)
+			}
+			for _, offset := range []time.Duration{-2 * time.Hour, -80 * time.Minute, -40 * time.Minute, -5 * time.Minute} {
+				failed(offset, models.MailboxNotLoadedPrefix)
+			}
+			failed(tc.boundary, "Other provider failure")
+			var first time.Time
+			if err := pool.QueryRow(t.Context(), `SELECT first_at FROM (`+warmupSendFailuresSQL+`) failures`, f.account, mailboxLoadingPattern, now).Scan(&first); err != nil {
+				t.Fatal(err)
+			}
+			if !first.Equal(now.Add(tc.first)) {
+				t.Fatalf("first failure=%v want=%v", first, now.Add(tc.first))
+			}
+		})
+	}
+}
+
 func TestLiveDispatchRetryDoesNotStarveNewerTask(t *testing.T) {
 	_, pool := liveContactDB(t)
 	ctx, now := context.Background(), time.Now().UTC().Truncate(time.Microsecond)