feat: record compatible warmup event schemas and keep unassigned filings from delaying recoverable mail

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

docs/content/docs/development/data-control.mdx

@@ -130,7 +130,7 @@ A [tracking comparison](/guides/placement-tests/#tracking) renders each seed's c
 
 `warmup_recovery_identifiers` holds SHA-256 fingerprints of a mailbox address and exact warmup message IDs or verification tokens, scoped to a workspace. They survive removing a mailbox so adding the same address again can recognize recent warmup mail without retaining its content or credentials. Each expires 90 days after its source record, expired identifiers are swept every minute in bounded batches, and deleting the workspace removes them immediately. They travel in the Warmup group of a [workspace export](/guides/workspace-export-import/).
 
-`warmup_pending_filings` holds message locators awaiting provider filing and worker acknowledgement. Failed or unassigned attempts retry after five minutes without replaying engagement signals. An acknowledgement or mailbox deletion removes the row. This instance-local queue does not travel in workspace exports; the destination resyncs the mailbox and files recognized warmup messages.
+`warmup_pending_filings` holds message locators awaiting provider filing and worker acknowledgement. Failed attempts retry after five minutes; unassigned mailboxes remain pending until reassignment, without replaying engagement signals. An acknowledgement or mailbox deletion removes the row. This instance-local queue does not travel in workspace exports; the destination resyncs the mailbox and files recognized warmup messages.
 
 ### Mailbox import files
 

internal/app/consumer/warmup_filing_recovery.go

@@ -51,7 +51,6 @@ func (s *JobsService) retryWarmupFilings(ctx context.Context) error {
 			failures = append(failures, s.WarmupRecoveryRepo.CompleteFiling(ctx, action.EmailID, id))
 			continue
 		}
-		s.markSelfMove(ctx, action.EmailID, action.RFCMessageID)
 		if err := s.Publisher.PublishWarmupAction(ctx, *account.WorkerID, &action); err != nil {
 			failures = append(failures, err)
 		}

internal/app/consumer/warmup_inbox_cleanup.go

@@ -62,10 +62,6 @@ func (s *JobsService) fileWarmupOutOfMailbox(ctx context.Context, e *models.JobE
 	if placement == models.WarmupPlacementInbox {
 		return nil
 	}
-	// Marked before publishing, like the live path: the move can land and be
-	// observed before a marker written afterwards would exist, and a mailbox
-	// must not be struck for foldering we asked it to do.
-	s.markSelfMove(ctx, e.Message.EmailID, e.Message.MessageID)
 	action := models.WarmupEmailAction{
 		UserID:             e.UserID,
 		EmailID:            e.Message.EmailID,
@@ -92,6 +88,10 @@ func (s *JobsService) fileWarmupOutOfMailbox(ctx context.Context, e *models.JobE
 		}
 		return fmt.Errorf("warmup filing: mailbox has no assigned worker")
 	}
+	if s.WarmupRecoveryRepo == nil {
+		s.markSelfMove(ctx, e.Message.EmailID, e.Message.MessageID)
+	}
+	// Durable filings use presence verification instead of assuming every attempt moves mail.
 	err := s.Publisher.PublishWarmupAction(ctx, *account.WorkerID, &action)
 	if s.WarmupRecoveryRepo != nil {
 		return nil // The persisted filing survives a failed publish.

internal/app/email/loader.go

@@ -259,7 +259,9 @@ func (s *emailService) dropFromWorker(ctx context.Context, userID string, accoun
 	}
 	if source, ok := s.emailRepository.(repository.WarmupDisconnectSource); ok {
 		account, aerr := s.emailRepository.GetByID(ctx, accountID)
-		if aerr == nil && account != nil {
+		if aerr != nil {
+			log.Warn().Err(aerr).Str("email_id", accountID.String()).Msg("pre-disconnect mailbox lookup failed")
+		} else if account != nil {
 			remove.WarmupPlacement, remove.WarmupFolder = account.WarmupFiling()
 			if remove.WarmupPlacement != models.WarmupPlacementInbox {
 				ids, err := source.WarmupDisconnectMessageIDs(ctx, accountID, 200)

internal/app/eventschemas/testdata/jobs.worker-events.avsc

@@ -887,6 +887,22 @@
           "name": "warmbly.events.JobEventTokenUpdate",
           "type": "record"
         },
+        {
+          "fields": [
+            {
+              "default": "",
+              "name": "email_id",
+              "type": "string"
+            },
+            {
+              "default": "",
+              "name": "filing_id",
+              "type": "string"
+            }
+          ],
+          "name": "warmbly.events.JobEventWarmupFiled",
+          "type": "record"
+        },
         {
           "fields": [
             {

internal/app/eventschemas/testdata/worker-commands.avsc

@@ -680,6 +680,24 @@
               "default": "",
               "name": "email_id",
               "type": "string"
+            },
+            {
+              "default": [],
+              "name": "warmup_message_ids",
+              "type": {
+                "items": "string",
+                "type": "array"
+              }
+            },
+            {
+              "default": "",
+              "name": "warmup_placement",
+              "type": "string"
+            },
+            {
+              "default": "",
+              "name": "warmup_folder",
+              "type": "string"
             }
           ],
           "name": "warmbly.events.RemoveWorkerEmail",
@@ -881,6 +899,11 @@
               "name": "internal_id",
               "type": "string"
             },
+            {
+              "default": "",
+              "name": "filing_id",
+              "type": "string"
+            },
             {
               "default": false,
               "name": "recheck",

internal/app/worker/event_remove_email.go

@@ -56,9 +56,7 @@ func (w *WorkerService) fileWarmupBeforeDisconnect(ctx context.Context, e *model
 		if ctx.Err() != nil {
 			break
 		}
-		if w.Cache != nil {
-			_ = w.Cache.Set(ctx, "warmup:selfmoved:"+e.EmailID+":"+messageID, "1", 30*time.Minute).Err()
-		}
+		// Provider removals use the normal mailbox-presence verification path.
 		if err := w.HandleWarmupAction(ctx, models.WarmupEmailAction{
 			UserID: userID, EmailID: accountID, RFCMessageID: messageID,
 			Placement: e.WarmupPlacement, TargetFolder: e.WarmupFolder,

internal/repository/pg_warmup_recovery.go

@@ -108,8 +108,10 @@ func (r *warmupRecoveryRepository) EnqueueFiling(ctx context.Context, action mod
 
 func (r *warmupRecoveryRepository) ClaimFilings(ctx context.Context, limit int) ([]models.WarmupEmailAction, error) {
 	rows, err := r.db.Query(ctx, `UPDATE warmup_pending_filings SET next_attempt_at = NOW() + INTERVAL '5 minutes'
-		WHERE id IN (SELECT id FROM warmup_pending_filings WHERE next_attempt_at <= NOW()
-		  ORDER BY next_attempt_at LIMIT $1 FOR UPDATE SKIP LOCKED)
+		WHERE id IN (SELECT f.id FROM warmup_pending_filings f
+		  JOIN email_accounts a ON a.id = f.email_account_id
+		  WHERE f.next_attempt_at <= NOW() AND a.worker_id IS NOT NULL
+		  ORDER BY f.next_attempt_at LIMIT $1 FOR UPDATE OF f SKIP LOCKED)
 		RETURNING id, payload`, limit)
 	if err != nil {
 		return nil, err

internal/repository/warmup_recovery_live_test.go

@@ -115,13 +115,24 @@ func TestLiveWarmupFilingRequiresAcknowledgement(t *testing.T) {
 	requireSchemaVersion(t, pool, 265)
 	ctx := context.Background()
 	f := newSharedOrgFixture(t, pool)
-	account := uuid.New()
+	account, waitingAccount, worker := uuid.New(), uuid.New(), uuid.New()
+	if _, err := pool.Exec(ctx, `INSERT INTO fleet_nodes (id, role, name) VALUES ($1, 'worker', 'filing-test')`, worker); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if _, err := pool.Exec(ctx, `DELETE FROM fleet_nodes WHERE id = $1`, worker); err != nil {
+			t.Error(err)
+		}
+	})
+	if _, err := pool.Exec(ctx, `INSERT INTO workers (id) VALUES ($1)`, worker); err != nil {
+		t.Fatal(err)
+	}
 	if _, err := pool.Exec(ctx, `INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, signature_html, provider, status)
 		VALUES ($1, $2, $3, $4, 'Filing', '', '', 'gmail', 'active')`, account, f.owner, f.org, uuid.NewString()+"@example.test"); err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		if _, err := pool.Exec(ctx, `DELETE FROM email_accounts WHERE id = $1`, account); err != nil {
+		if _, err := pool.Exec(ctx, `DELETE FROM email_accounts WHERE id = ANY($1)`, []uuid.UUID{account, waitingAccount}); err != nil {
 			t.Error(err)
 		}
 	})
@@ -136,13 +147,36 @@ func TestLiveWarmupFilingRequiresAcknowledgement(t *testing.T) {
 	if err != nil || id != duplicate {
 		t.Fatalf("duplicate filing: %v, %v, %v", id, duplicate, err)
 	}
-	claimed, err := repo.ClaimFilings(ctx, 100)
+	if claimed, err := repo.ClaimFilings(ctx, 1); err != nil || len(claimed) != 0 {
+		t.Fatalf("unassigned filing consumed a claim: %+v %v", claimed, err)
+	}
+	if _, err := pool.Exec(ctx, `INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, signature_html, provider, status)
+		VALUES ($1, $2, $3, $4, 'Waiting', '', '', 'gmail', 'active')`, waitingAccount, f.owner, f.org, uuid.NewString()+"@example.test"); err != nil {
+		t.Fatal(err)
+	}
+	waitingAction := action
+	waitingAction.EmailID = waitingAccount
+	waitingID, err := repo.EnqueueFiling(ctx, waitingAction)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE warmup_pending_filings SET next_attempt_at = NOW() - INTERVAL '1 hour' WHERE id = $1`, waitingID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE email_accounts SET worker_id = $2 WHERE id = $1`, account, worker); err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := repo.ClaimFilings(ctx, 1)
 	if err != nil || len(claimed) != 1 || claimed[0].FilingID != id.String() || len(claimed[0].Actions) != 1 {
 		t.Fatalf("claim: %+v %v", claimed, err)
 	}
 	if claimed, err := repo.ClaimFilings(ctx, 100); err != nil || len(claimed) != 0 {
 		t.Fatalf("lease did not prevent duplicate dispatch: %+v %v", claimed, err)
 	}
+	var waiting bool
+	if err := pool.QueryRow(ctx, `SELECT next_attempt_at < NOW() - INTERVAL '30 minutes' FROM warmup_pending_filings WHERE id = $1`, waitingID).Scan(&waiting); err != nil || !waiting {
+		t.Fatalf("unassigned backlog was leased instead of remaining pending: %v %v", waiting, err)
+	}
 	if err := repo.CompleteFiling(ctx, uuid.New(), id); err != nil {
 		t.Fatal(err)
 	}