Merge pull request #927 from warmbly/devin/1791632533-isolate-warmup-provider-retries

feat: defer failed warmup filings only with authoritative durable recovery proof

.github/workflows/ci.yml

@@ -207,7 +207,7 @@ jobs:
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
         run: |
-          go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement|WarmupTimeWindow|WarmupInvalidWindow)' -count=1
+          go test -race ./internal/repository -run '^TestLive(Monitoring|LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|AdminPlatformOverview|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains|WarmupFilingRequiresAcknowledgement|WarmupFilingAuthority|WarmupTimeWindow|WarmupInvalidWindow)' -count=1
           go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults|DispatchDefersInvalidHours)|WarmupSenderDistinguishes)' -count=1
           go test -race ./internal/app/consumer -run '^TestLiveStoredSendResult' -count=1
 

docs/content/docs/api/endpoints.mdx

@@ -643,6 +643,8 @@ Cloud-managed OAuth additionally exposes `POST /pool-link/instance/oauth/start-c
 
 Warmup actions for a deleted, unassigned or no-longer-authorized mailbox return no permitted actions. The worker treats that authoritative denial as a no-op without touching the provider. Database or authority failures still return `503` and retry; denial does not authorize a send, clear a send hold, or acknowledge an arrival.
 
+`POST /internal/worker/warmup-actions` optionally accepts `filing_id` for a filing-only action. `filing_pending: true` is returned only for permitted, pure filing work still durably queued for that exact mailbox and its current worker. On a failed filing, this proof lets an updated worker finish the command without a `WarmupFiled` completion event; the durable recovery queue remains responsible for retrying it. Missing proof or an older backend retains the existing command retries. Deploy the backend before workers to enable this behavior. Successful filing still requires a completion event, and publication failures remain retryable. No send, arrival or retention-delete authority is added.
+
 ### Referrals and discounts
 
 The referral program and billing discount history are part of `/subscription/*`, so they are JWT only and never accept an API key. There is no API permission scope for them; browser callers are gated by the org permission below. The discount validation and checkout endpoints are listed under [Subscription and billing](/api/reference/account-org/) in the reference.

docs/content/docs/guides/warmup.mdx

@@ -272,6 +272,8 @@ Mail that arrives somewhere unexpected is still handled. A server-side rule, or
 
 Filing happens as each message arrives, so changing the setting applies to warmup from that point on.
 
+If filing fails, it is not reported as completed. An updated worker can release that filing command for later retry only after the backend confirms that the exact filing is still in its durable recovery queue. Recovery retries use the current worker assignment and folder setting, spaced by 5, 15 or 30 minutes as the queued filing ages. Unknown recovery authority keeps the existing command retry behavior. This does not resend mail, clear uncertain-send protection, acknowledge mailbox arrivals, or discard retention deletes and removal checks.
+
 A sweep identifies missed application-owned diagnostics and applies authorized housekeeping. It does not establish retroactive authentication or engagement. Leaving mail in the inbox remains an owner choice.
 
 The sweep repairs what it can still see, so it does not clear a backlog that built up before the setting existed. To do that on Gmail, search `label:Warmbly` and archive the results: warmup mail has always carried the label, so that selects exactly it. On Outlook and IMAP the received mail is already in the folder, and only the Sent copies from before this change stay in Sent.

internal/api/handler/internal_warmup_actions.go

@@ -3,20 +3,32 @@ package handler
 import (
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 	"net/http"
 )
 
 func (h *Handler) InternalWarmupActions(c *gin.Context) {
-	var req struct {
-		MailboxID uuid.UUID `json:"mailbox_id"`
-		WorkerID  uuid.UUID `json:"worker_id"`
-		Actions   []string  `json:"actions"`
-	}
+	var req models.WarmupActionRequest
 	if err := c.ShouldBindJSON(&req); err != nil || req.MailboxID == uuid.Nil || req.WorkerID == uuid.Nil || len(req.Actions) > 16 {
 		c.Status(http.StatusBadRequest)
 		return
 	}
+	if req.FilingID != "" {
+		if id, err := uuid.Parse(req.FilingID); err != nil || id == uuid.Nil {
+			c.Status(http.StatusBadRequest)
+			return
+		}
+	}
+	if r, ok := h.WarmupDispatch.(repository.WarmupActionRecoveryAdmission); ok {
+		decision, err := r.AdmitWarmupAction(c.Request.Context(), req)
+		if err != nil {
+			c.Status(http.StatusServiceUnavailable)
+			return
+		}
+		c.JSON(http.StatusOK, decision)
+		return
+	}
 	r, ok := h.WarmupDispatch.(repository.WarmupActionAdmission)
 	if !ok {
 		c.Status(http.StatusServiceUnavailable)

internal/api/handler/internal_warmup_actions_test.go

@@ -0,0 +1,89 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+type warmupRecoveryAuthority struct {
+	repository.WarmupDispatchRepository
+	request models.WarmupActionRequest
+	err     error
+}
+
+func (r *warmupRecoveryAuthority) AdmitWarmupAction(_ context.Context, req models.WarmupActionRequest) (models.WarmupActionDecision, error) {
+	r.request = req
+	return models.WarmupActionDecision{Actions: req.Actions, FilingPending: req.FilingID != ""}, r.err
+}
+
+type legacyWarmupAuthority struct {
+	repository.WarmupDispatchRepository
+}
+
+func (r *legacyWarmupAuthority) PermittedWarmupActions(_ context.Context, _, _ uuid.UUID, actions []string) ([]string, error) {
+	return actions, nil
+}
+
+func TestInternalWarmupFilingAdmissionCompatibility(t *testing.T) {
+	request := models.WarmupActionRequest{MailboxID: uuid.New(), WorkerID: uuid.New(), Actions: []string{models.WarmupActionFile}, FilingID: uuid.NewString()}
+	for _, tc := range []struct {
+		name      string
+		authority repository.WarmupDispatchRepository
+		status    int
+		pending   bool
+	}{
+		{"durable authority", &warmupRecoveryAuthority{}, http.StatusOK, true},
+		{"legacy authority", &legacyWarmupAuthority{}, http.StatusOK, false},
+		{"failed authority", &warmupRecoveryAuthority{err: errors.New("database unavailable")}, http.StatusServiceUnavailable, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			body, _ := json.Marshal(request)
+			out := httptest.NewRecorder()
+			httpRequest := httptest.NewRequest(http.MethodPost, "/internal/worker/warmup-actions", strings.NewReader(string(body)))
+			httpRequest.Header.Set("Content-Type", "application/json")
+			h := Handler{WarmupDispatch: tc.authority}
+			router := gin.New()
+			router.POST("/internal/worker/warmup-actions", h.InternalWarmupActions)
+			router.ServeHTTP(out, httpRequest)
+			if out.Code != tc.status {
+				t.Fatalf("status=%d body=%s", out.Code, out.Body.String())
+			}
+			if out.Code == http.StatusOK {
+				var decision models.WarmupActionDecision
+				if err := json.Unmarshal(out.Body.Bytes(), &decision); err != nil || decision.FilingPending != tc.pending || len(decision.Actions) != 1 {
+					t.Fatalf("decision=%+v err=%v", decision, err)
+				}
+			}
+			if r, ok := tc.authority.(*warmupRecoveryAuthority); ok && r.request.FilingID != request.FilingID {
+				t.Fatal("filing binding lost")
+			}
+		})
+	}
+}
+
+func TestInternalWarmupFilingAdmissionRejectsInvalidIdentifier(t *testing.T) {
+	for _, id := range []string{"invalid", uuid.Nil.String()} {
+		r := &warmupRecoveryAuthority{}
+		body, _ := json.Marshal(models.WarmupActionRequest{MailboxID: uuid.New(), WorkerID: uuid.New(), Actions: []string{models.WarmupActionFile}, FilingID: id})
+		out := httptest.NewRecorder()
+		request := httptest.NewRequest(http.MethodPost, "/internal/worker/warmup-actions", strings.NewReader(string(body)))
+		request.Header.Set("Content-Type", "application/json")
+		h := Handler{WarmupDispatch: r}
+		router := gin.New()
+		router.POST("/internal/worker/warmup-actions", h.InternalWarmupActions)
+		router.ServeHTTP(out, request)
+		if out.Code != http.StatusBadRequest || r.request.FilingID != "" {
+			t.Fatal("invalid filing reached recovery authority")
+		}
+	}
+}

internal/app/worker/event_warmup_action.go

@@ -27,12 +27,18 @@ import (
 // DelaySeconds=0. That makes the dwell survive a worker restart, which the old
 // in-process time.AfterFunc here could not.
 func (w *WorkerService) HandleWarmupAction(ctx context.Context, action models.WarmupEmailAction) error {
+	filingPending := false
 	if gate, ok := w.SyncContextRepository.(repository.WarmupActionAdmission); ok {
 		worker, err := uuid.Parse(w.ID)
 		if err != nil {
 			return err
 		}
-		action.Actions, err = gate.PermittedWarmupActions(ctx, action.EmailID, worker, action.Actions)
+		if recovery, ok := w.SyncContextRepository.(repository.WarmupActionRecoveryAdmission); ok && action.FilingID != "" && len(action.Actions) == 1 && action.Actions[0] == models.WarmupActionFile {
+			decision, xerr := recovery.AdmitWarmupAction(ctx, models.WarmupActionRequest{MailboxID: action.EmailID, WorkerID: worker, Actions: action.Actions, FilingID: action.FilingID})
+			action.Actions, filingPending, err = decision.Actions, decision.FilingPending, xerr
+		} else {
+			action.Actions, err = gate.PermittedWarmupActions(ctx, action.EmailID, worker, action.Actions)
+		}
 		if err != nil {
 			return err
 		}
@@ -101,6 +107,12 @@ func (w *WorkerService) HandleWarmupAction(ctx context.Context, action models.Wa
 		}
 		return nil
 	}
+	if filingPending && len(action.Actions) == 1 && action.Actions[0] == models.WarmupActionFile {
+		// Only the confirmed durable filing queue can take over this failed command.
+		log.Warn().Err(err).Str("email_id", action.EmailID.String()).Str("filing_id", action.FilingID).
+			Msg("Warmup filing deferred to durable recovery")
+		return nil
+	}
 	// Engagement is best effort and never returns here. A retention delete
 	// (the row is already retired) and a removal check are redelivered, a
 	// bounded number of times.

internal/app/worker/event_warmup_action_test.go

@@ -17,6 +17,61 @@ type warmupActionAuthorityStub struct {
 	err error
 }
 
+type warmupFilingAuthorityStub struct {
+	repository.SyncContextRepository
+	pending  bool
+	err      error
+	requests []models.WarmupActionRequest
+}
+
+func (s *warmupFilingAuthorityStub) PermittedWarmupActions(_ context.Context, _, _ uuid.UUID, actions []string) ([]string, error) {
+	return actions, s.err
+}
+
+func (s *warmupFilingAuthorityStub) AdmitWarmupAction(_ context.Context, req models.WarmupActionRequest) (models.WarmupActionDecision, error) {
+	s.requests = append(s.requests, req)
+	return models.WarmupActionDecision{Actions: req.Actions, FilingPending: s.pending}, s.err
+}
+
+func TestDurableWarmupFilingDefersOnlyWithPositiveAuthority(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		pending      bool
+		authorityErr error
+		actions      []string
+		filing       bool
+		deferred     bool
+	}{
+		{"confirmed filing", true, nil, []string{models.WarmupActionFile}, true, true},
+		{"absent proof", false, nil, []string{models.WarmupActionFile}, true, false},
+		{"authority unavailable", true, errors.New("authority unavailable"), []string{models.WarmupActionFile}, true, false},
+		{"mixed deletion", true, nil, []string{models.WarmupActionFile, models.WarmupActionDelete}, true, false},
+		{"retention delete", true, nil, []string{models.WarmupActionDelete}, false, false},
+		{"removal check", true, nil, []string{models.WarmupActionVerifyRemoval}, false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			account := uuid.New()
+			w, bus := filingWorker(account, &filingIMAP{found: map[string]uint32{"INBOX": 11}, err: errors.New("provider unavailable")})
+			w.ID = uuid.NewString()
+			delete(w.mailManager.Emails, account)
+			authority := &warmupFilingAuthorityStub{pending: tc.pending, err: tc.authorityErr}
+			w.SyncContextRepository = authority
+			action := models.WarmupEmailAction{EmailID: account, Actions: tc.actions}
+			if tc.filing {
+				action.FilingID = uuid.NewString()
+			}
+			ctx := context.WithValue(t.Context(), deliveryKey{}, delivery{attempt: 1, redelivers: true})
+			err := w.HandleWarmupAction(ctx, action)
+			if (err == nil) != tc.deferred || len(bus.events) != 0 {
+				t.Fatalf("deferred=%v error=%v events=%v", tc.deferred, err, bus.events)
+			}
+			if tc.deferred && (len(authority.requests) != 1 || authority.requests[0].FilingID != action.FilingID || authority.requests[0].MailboxID != account || authority.requests[0].WorkerID.String() != w.ID) {
+				t.Fatal("durable proof was not bound to the exact mailbox, worker and filing")
+			}
+		})
+	}
+}
+
 func (s *warmupActionAuthorityStub) PermittedWarmupActions(context.Context, uuid.UUID, uuid.UUID, []string) ([]string, error) {
 	return nil, s.err
 }

internal/app/worker/warmup_filing_recovery_test.go

@@ -237,3 +237,68 @@ func TestDurableGraphFilingRetriesWithoutRestoringTrashedMail(t *testing.T) {
 		t.Fatal("Graph filing restored a message from Trash or failed to resolve the pending filing")
 	}
 }
+
+func TestDurableGraphFilingFailureReleasesCommandWithoutFilingAcknowledgement(t *testing.T) {
+	for _, stage := range []string{"locate", "move"} {
+		t.Run(stage, func(t *testing.T) {
+			failing := true
+			transport := filingTransport(func(r *http.Request) (*http.Response, error) {
+				body, status := `{}`, http.StatusOK
+				switch {
+				case strings.HasSuffix(r.URL.Path, "/messages"):
+					body = `{"value":[{"id":"resolved"}]}`
+					if failing && stage == "locate" {
+						status = http.StatusServiceUnavailable
+					}
+				case strings.HasSuffix(r.URL.Path, "/mailFolders/deleteditems"):
+					body = `{"id":"trash-id"}`
+				case strings.HasSuffix(r.URL.Path, "/mailFolders"):
+					body = `{"value":[{"id":"target-id","displayName":"Warmbly"}]}`
+				case strings.HasSuffix(r.URL.Path, "/messages/resolved"):
+					body = `{"parentFolderId":"inbox-id"}`
+				case strings.HasSuffix(r.URL.Path, "/messages/resolved/move"):
+					if failing {
+						status = http.StatusServiceUnavailable
+					}
+					body = `{"id":"resolved"}`
+				default:
+					t.Errorf("unexpected Graph request: %s", r.URL.Path)
+				}
+				if status == http.StatusServiceUnavailable {
+					body = `{"error":{"code":"Unavailable","message":"provider unavailable"}}`
+				}
+				return &http.Response{StatusCode: status, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header), Request: r}, nil
+			})
+			ctx := context.WithValue(t.Context(), oauth2.HTTPClient, &http.Client{Transport: transport})
+			client := &msgraph.Client{}
+			if err := client.Init(ctx, &oauth2.Token{AccessToken: "test", Expiry: time.Now().Add(time.Hour)}, oauth2.Config{}); err != nil {
+				t.Fatal(err)
+			}
+			account := uuid.New()
+			w, bus := filingWorker(account, &filingIMAP{})
+			w.ID = uuid.NewString()
+			w.mailManager.Emails[account] = &wmail.WMail{GraphData: &wmail.GraphData{Client: client}}
+			authority := &warmupFilingAuthorityStub{pending: true}
+			w.SyncContextRepository = authority
+			action := models.WarmupEmailAction{EmailID: account, FilingID: uuid.NewString(), RFCMessageID: "<warmup@example.test>", Actions: []string{models.WarmupActionFile}}
+			ctx = context.WithValue(ctx, deliveryKey{}, delivery{attempt: 1, redelivers: true})
+			if err := w.HandleWarmupAction(ctx, action); err != nil || len(bus.events) != 0 {
+				t.Fatalf("durable Graph failure blocked the command or falsely acknowledged filing: %v %+v", err, bus.events)
+			}
+			// A healthy mailbox can run before the control plane retries the failed filing.
+			other := uuid.New()
+			w.mailManager.Emails[other] = &wmail.WMail{SmtpImapData: &wmail.SmtpImapData{ImapClient: &filingIMAP{found: map[string]uint32{"INBOX": 12}}, Mailboxes: []*models.Mailbox{{Name: "INBOX", UIDValidity: 77}}}}
+			if err := w.HandleWarmupAction(ctx, models.WarmupEmailAction{EmailID: other, FilingID: uuid.NewString(), RFCMessageID: "<other@example.test>", Actions: []string{models.WarmupActionFile}}); err != nil || len(bus.events) != 1 {
+				t.Fatalf("healthy filing blocked: %v", err)
+			}
+			failing = false
+			if err := w.HandleWarmupAction(ctx, action); err != nil || len(bus.events) != 2 {
+				t.Fatalf("durable retry did not acknowledge success: %v %+v", err, bus.events)
+			}
+			bus.err = errors.New("result broker unavailable")
+			if err := w.HandleWarmupAction(ctx, action); err == nil {
+				t.Fatal("filing acknowledgement failure was discarded")
+			}
+		})
+	}
+}

internal/models/warmup.go

@@ -78,6 +78,18 @@ const (
 	WarmupRemovalUnknown = "unknown"
 )
 
+type WarmupActionRequest struct {
+	MailboxID uuid.UUID `json:"mailbox_id"`
+	WorkerID  uuid.UUID `json:"worker_id"`
+	Actions   []string  `json:"actions"`
+	FilingID  string    `json:"filing_id,omitempty"`
+}
+
+type WarmupActionDecision struct {
+	Actions       []string `json:"actions"`
+	FilingPending bool     `json:"filing_pending,omitempty"`
+}
+
 // WarmupEmailAction represents actions to perform on a detected warmup email.
 //
 // For Gmail accounts the worker uses GmailID to issue Users.Messages.Modify

internal/repository/test_participation_live_test.go

@@ -45,6 +45,88 @@ func TestLiveWarmupActionsPreserveAuthorityFailure(t *testing.T) {
 	}
 }
 
+func TestLiveWarmupFilingAuthorityRequiresOwnedPendingWork(t *testing.T) {
+	f, r := lineageFixture(t)
+	worker := uuid.New()
+	ctx := t.Context()
+	exec := func(query string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, query, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol)VALUES($1,'worker',true,NOW(),2)`, worker)
+	exec(`INSERT INTO workers(id)VALUES($1)`, worker)
+	t.Cleanup(func() { _, _ = f.pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+	exec(`UPDATE email_accounts SET worker_id=$1,test_mode='diagnostic',test_receive_enabled=true,send_recovery_hold=true,send_recovery_reason='unknown' WHERE id=$2`, worker, f.recipient)
+	recovery := NewWarmupRecoveryRepository(f.pool)
+	id, err := recovery.EnqueueFiling(ctx, models.WarmupEmailAction{EmailID: f.recipient, RFCMessageID: "<durable@example.test>", Actions: []string{models.WarmupActionFile}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	req := models.WarmupActionRequest{MailboxID: f.recipient, WorkerID: worker, FilingID: id.String(), Actions: []string{models.WarmupActionFile}}
+	for _, tc := range []struct {
+		name    string
+		change  func(*models.WarmupActionRequest)
+		pending bool
+	}{
+		{"pending", func(*models.WarmupActionRequest) {}, true},
+		{"missing filing", func(r *models.WarmupActionRequest) { r.FilingID = uuid.NewString() }, false},
+		{"other mailbox", func(r *models.WarmupActionRequest) { r.MailboxID = f.sender }, false},
+		{"other worker", func(r *models.WarmupActionRequest) { r.WorkerID = uuid.New() }, false},
+		{"mixed action", func(r *models.WarmupActionRequest) {
+			r.Actions = []string{models.WarmupActionFile, models.WarmupActionDelete}
+		}, false},
+		{"legacy event", func(r *models.WarmupActionRequest) { r.FilingID = "" }, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			patch := req
+			tc.change(&patch)
+			out, err := r.AdmitWarmupAction(ctx, patch)
+			if err != nil || out.FilingPending != tc.pending {
+				t.Fatalf("decision=%+v err=%v", out, err)
+			}
+		})
+	}
+	exec(`UPDATE warmup_pending_filings SET payload=jsonb_set(payload,'{actions}',to_jsonb($2::text[])) WHERE id=$1`, id, []string{models.WarmupActionFile, models.WarmupActionDelete})
+	if out, err := r.AdmitWarmupAction(ctx, req); err != nil || out.FilingPending {
+		t.Fatal("mixed durable work granted filing-only proof", out, err)
+	}
+	exec(`UPDATE warmup_pending_filings SET payload=jsonb_set(payload,'{actions}',to_jsonb($2::text[])) WHERE id=$1`, id, []string{models.WarmupActionFile})
+	// Inspecting authority does not complete work or bypass recovery backoff.
+	claimed, err := recovery.ClaimFilings(ctx, 1)
+	if err != nil || len(claimed) != 1 || claimed[0].FilingID != id.String() {
+		t.Fatal("filing was lost before acknowledgement", claimed, err)
+	}
+	if claimed, err := recovery.ClaimFilings(ctx, 1); err != nil || len(claimed) != 0 {
+		t.Fatal("retry backoff was lost", claimed, err)
+	}
+	if out, err := r.AdmitWarmupAction(ctx, req); err != nil || !out.FilingPending {
+		t.Fatal("backoff erased durable authority", out, err)
+	}
+	exec(`UPDATE email_accounts SET test_mode='off' WHERE id=$1`, f.recipient)
+	if out, err := r.AdmitWarmupAction(ctx, req); err != nil || out.FilingPending || len(out.Actions) != 0 {
+		t.Fatal("revoked filing remained permitted", out, err)
+	}
+	exec(`UPDATE email_accounts SET test_mode='diagnostic' WHERE id=$1`, f.recipient)
+	if err := recovery.CompleteFiling(ctx, f.recipient, id); err != nil {
+		t.Fatal(err)
+	}
+	if out, err := r.AdmitWarmupAction(ctx, req); err != nil || out.FilingPending {
+		t.Fatal("completed filing retained pending proof", out, err)
+	}
+	var held bool
+	var reason string
+	if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_reason FROM email_accounts WHERE id=$1`, f.recipient).Scan(&held, &reason); err != nil || !held || reason != "unknown" {
+		t.Fatal("filing authority changed uncertain-send protection", err)
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	cancel()
+	if _, err := r.AdmitWarmupAction(ctx, req); !errors.Is(err, context.Canceled) {
+		t.Fatal("authority failure did not remain retryable", err)
+	}
+}
+
 func TestLiveDiagnosticOffStopsQueuedSendAndActions(t *testing.T) {
 	f, r := lineageFixture(t)
 	ctx := t.Context()

internal/repository/warmup_action_admission.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"io"
 	"net/http"
 
 	"github.com/google/uuid"
@@ -16,6 +17,27 @@ type WarmupActionAdmission interface {
 	PermittedWarmupActions(context.Context, uuid.UUID, uuid.UUID, []string) ([]string, error)
 }
 
+type WarmupActionRecoveryAdmission interface {
+	AdmitWarmupAction(context.Context, models.WarmupActionRequest) (models.WarmupActionDecision, error)
+}
+
+func (r *taskRepository) AdmitWarmupAction(ctx context.Context, req models.WarmupActionRequest) (models.WarmupActionDecision, error) {
+	actions, err := r.PermittedWarmupActions(ctx, req.MailboxID, req.WorkerID, req.Actions)
+	out := models.WarmupActionDecision{Actions: actions}
+	if err != nil || req.FilingID == "" || len(req.Actions) != 1 || req.Actions[0] != models.WarmupActionFile || len(actions) != 1 || actions[0] != models.WarmupActionFile {
+		return out, err
+	}
+	filing, err := uuid.Parse(req.FilingID)
+	if err != nil || filing == uuid.Nil {
+		return out, errors.New("invalid warmup filing identifier")
+	}
+	err = r.db.QueryRow(ctx, `SELECT EXISTS (
+		SELECT 1 FROM warmup_pending_filings f JOIN email_accounts ea ON ea.id=f.email_account_id
+		WHERE f.id=$3 AND f.email_account_id=$1 AND ea.worker_id=$2 AND f.payload->'actions'=to_jsonb($4::text[])
+	)`, req.MailboxID, req.WorkerID, filing, []string{models.WarmupActionFile}).Scan(&out.FilingPending)
+	return out, err
+}
+
 func (r *taskRepository) PermittedWarmupActions(ctx context.Context, mailbox, worker uuid.UUID, actions []string) ([]string, error) {
 	if len(actions) > 16 {
 		return nil, ErrSendAdmissionDenied
@@ -38,31 +60,45 @@ func (r *taskRepository) PermittedWarmupActions(ctx context.Context, mailbox, wo
 }
 
 func (r *httpSyncContextRepository) PermittedWarmupActions(ctx context.Context, mailbox, worker uuid.UUID, actions []string) ([]string, error) {
-	raw, err := json.Marshal(struct {
-		MailboxID uuid.UUID `json:"mailbox_id"`
-		WorkerID  uuid.UUID `json:"worker_id"`
-		Actions   []string  `json:"actions"`
-	}{mailbox, worker, actions})
+	out, err := r.AdmitWarmupAction(ctx, models.WarmupActionRequest{MailboxID: mailbox, WorkerID: worker, Actions: actions})
+	return out.Actions, err
+}
+
+func (r *httpSyncContextRepository) AdmitWarmupAction(ctx context.Context, request models.WarmupActionRequest) (models.WarmupActionDecision, error) {
+	var out models.WarmupActionDecision
+	raw, err := json.Marshal(request)
 	if err != nil {
-		return nil, err
+		return out, err
 	}
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.baseURL+"/api/v1/internal/worker/warmup-actions", bytes.NewReader(raw))
 	if err != nil {
-		return nil, err
+		return out, err
 	}
 	req.Header.Set("Authorization", "Bearer "+r.token)
 	req.Header.Set("Content-Type", "application/json")
 	resp, err := r.client.Do(req)
 	if err != nil {
-		return nil, err
+		return out, err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusOK {
-		return nil, errors.New("warmup action authority unavailable")
+		return out, errors.New("warmup action authority unavailable")
 	}
-	var out struct {
-		Actions []string `json:"actions"`
+	var response struct {
+		Actions       json.RawMessage `json:"actions"`
+		FilingPending bool            `json:"filing_pending"`
 	}
-	err = json.NewDecoder(resp.Body).Decode(&out)
-	return out.Actions, err
+	decoder := json.NewDecoder(resp.Body)
+	if err := decoder.Decode(&response); err != nil {
+		return out, err
+	}
+	var trailing any
+	if len(response.Actions) == 0 || decoder.Decode(&trailing) != io.EOF {
+		return out, errors.New("invalid warmup action authority response")
+	}
+	if err := json.Unmarshal(response.Actions, &out.Actions); err != nil {
+		return out, err
+	}
+	out.FilingPending = response.FilingPending
+	return out, nil
 }

internal/repository/warmup_action_admission_test.go

@@ -0,0 +1,54 @@
+package repository
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func TestWarmupFilingHTTPAuthorityCompatibility(t *testing.T) {
+	request := models.WarmupActionRequest{MailboxID: uuid.New(), WorkerID: uuid.New(), Actions: []string{models.WarmupActionFile}, FilingID: uuid.NewString()}
+	for _, tc := range []struct {
+		name, body       string
+		status           int
+		pending, failure bool
+	}{
+		{"confirmed", `{"actions":["move_to_warmbly"],"filing_pending":true}`, 200, true, false},
+		{"older backend", `{"actions":["move_to_warmbly"]}`, 200, false, false},
+		{"not pending", `{"actions":["move_to_warmbly"],"filing_pending":false}`, 200, false, false},
+		{"authority error", `{"actions":["move_to_warmbly"],"filing_pending":true}`, 503, false, true},
+		{"invalid response", `{"filing_pending":true,"actions":`, 200, false, true},
+		{"missing actions", `{"filing_pending":true}`, 200, false, true},
+		{"trailing garbage", `{"actions":["move_to_warmbly"],"filing_pending":true} invalid`, 200, false, true},
+		{"multiple responses", `{"actions":["move_to_warmbly"],"filing_pending":true}{}`, 200, false, true},
+		{"null response", `null`, 200, false, true},
+		{"denied actions", `{"actions":null}`, 200, false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var got models.WarmupActionRequest
+				if err := json.NewDecoder(r.Body).Decode(&got); err != nil {
+					t.Error(err)
+				}
+				if got.MailboxID != request.MailboxID || got.WorkerID != request.WorkerID || got.FilingID != request.FilingID || r.URL.Path != "/api/v1/internal/worker/warmup-actions" || r.Header.Get("Authorization") != "Bearer test" {
+					t.Error("authority request lost binding or authentication")
+				}
+				w.WriteHeader(tc.status)
+				_, _ = w.Write([]byte(tc.body))
+			}))
+			defer srv.Close()
+			r, err := NewHTTPSyncContextRepository(srv.URL, "test")
+			if err != nil {
+				t.Fatal(err)
+			}
+			out, err := r.(WarmupActionRecoveryAdmission).AdmitWarmupAction(t.Context(), request)
+			if (err != nil) != tc.failure || !tc.failure && out.FilingPending != tc.pending {
+				t.Fatalf("pending=%v err=%v", out.FilingPending, err)
+			}
+		})
+	}
+}