From 9a4525952748b3c45b9da827b2acbd336e487dca Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 11:52:47 +0000 Subject: [PATCH 1/2] feat: release failed filing commands only after exact backend proof of pending durable recovery while preserving acknowledgements and legacy retries --- .github/workflows/ci.yml | 2 +- docs/content/docs/api/endpoints.mdx | 2 + docs/content/docs/guides/warmup.mdx | 2 + .../api/handler/internal_warmup_actions.go | 22 +++-- .../handler/internal_warmup_actions_test.go | 89 +++++++++++++++++++ internal/app/worker/event_warmup_action.go | 14 ++- .../app/worker/event_warmup_action_test.go | 55 ++++++++++++ .../app/worker/warmup_filing_recovery_test.go | 65 ++++++++++++++ internal/models/warmup.go | 12 +++ .../test_participation_live_test.go | 82 +++++++++++++++++ .../repository/warmup_action_admission.go | 46 +++++++--- .../warmup_action_admission_test.go | 49 ++++++++++ 12 files changed, 420 insertions(+), 20 deletions(-) create mode 100644 internal/api/handler/internal_warmup_actions_test.go create mode 100644 internal/repository/warmup_action_admission_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cc94b110c..a561bca00 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index b502f6d9b..4596fd8a0 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/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. diff --git a/docs/content/docs/guides/warmup.mdx b/docs/content/docs/guides/warmup.mdx index 5365eb53b..1ab8bdbea 100644 --- a/docs/content/docs/guides/warmup.mdx +++ b/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. diff --git a/internal/api/handler/internal_warmup_actions.go b/internal/api/handler/internal_warmup_actions.go index 108fd67c8..a3d429db5 100644 --- a/internal/api/handler/internal_warmup_actions.go +++ b/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) diff --git a/internal/api/handler/internal_warmup_actions_test.go b/internal/api/handler/internal_warmup_actions_test.go new file mode 100644 index 000000000..7f33c698f --- /dev/null +++ b/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") + } + } +} diff --git a/internal/app/worker/event_warmup_action.go b/internal/app/worker/event_warmup_action.go index 8cc7ba62f..60eda4ed2 100644 --- a/internal/app/worker/event_warmup_action.go +++ b/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. diff --git a/internal/app/worker/event_warmup_action_test.go b/internal/app/worker/event_warmup_action_test.go index ae61870c8..ef9e4ebab 100644 --- a/internal/app/worker/event_warmup_action_test.go +++ b/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 } diff --git a/internal/app/worker/warmup_filing_recovery_test.go b/internal/app/worker/warmup_filing_recovery_test.go index a9b9fef69..d07a4389a 100644 --- a/internal/app/worker/warmup_filing_recovery_test.go +++ b/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: "", 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: "", 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") + } + }) + } +} diff --git a/internal/models/warmup.go b/internal/models/warmup.go index 74d337e20..c7aef9c06 100644 --- a/internal/models/warmup.go +++ b/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 diff --git a/internal/repository/test_participation_live_test.go b/internal/repository/test_participation_live_test.go index 90c683824..7d5d71fae 100644 --- a/internal/repository/test_participation_live_test.go +++ b/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: "", 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() diff --git a/internal/repository/warmup_action_admission.go b/internal/repository/warmup_action_admission.go index 769c43eb6..7be46b187 100644 --- a/internal/repository/warmup_action_admission.go +++ b/internal/repository/warmup_action_admission.go @@ -16,6 +16,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 +59,30 @@ 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") - } - var out struct { - Actions []string `json:"actions"` + return out, errors.New("warmup action authority unavailable") } err = json.NewDecoder(resp.Body).Decode(&out) - return out.Actions, err + return out, err } diff --git a/internal/repository/warmup_action_admission_test.go b/internal/repository/warmup_action_admission_test.go new file mode 100644 index 000000000..5532335f8 --- /dev/null +++ b/internal/repository/warmup_action_admission_test.go @@ -0,0 +1,49 @@ +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}, + } { + 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) + } + }) + } +} From fc3f1de9776fb42027a0ad96c7a90388a992c348 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 10 Oct 2026 11:55:16 +0000 Subject: [PATCH 2/2] feat: reject incomplete or trailing warmup authority responses before granting durable filing deferral --- .../repository/warmup_action_admission.go | 20 +++++++++++++++++-- .../warmup_action_admission_test.go | 5 +++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/internal/repository/warmup_action_admission.go b/internal/repository/warmup_action_admission.go index 7be46b187..ff417c7a2 100644 --- a/internal/repository/warmup_action_admission.go +++ b/internal/repository/warmup_action_admission.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "io" "net/http" "github.com/google/uuid" @@ -83,6 +84,21 @@ func (r *httpSyncContextRepository) AdmitWarmupAction(ctx context.Context, reque if resp.StatusCode != http.StatusOK { return out, errors.New("warmup action authority unavailable") } - err = json.NewDecoder(resp.Body).Decode(&out) - return out, err + var response struct { + Actions json.RawMessage `json:"actions"` + FilingPending bool `json:"filing_pending"` + } + 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 } diff --git a/internal/repository/warmup_action_admission_test.go b/internal/repository/warmup_action_admission_test.go index 5532335f8..26645fdf6 100644 --- a/internal/repository/warmup_action_admission_test.go +++ b/internal/repository/warmup_action_admission_test.go @@ -22,6 +22,11 @@ func TestWarmupFilingHTTPAuthorityCompatibility(t *testing.T) { {"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) {