feat: retire stale worker mailbox generations only on authenticated ownership-loss admission evidence while retaining failed arrival bodies and provider cursors

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

docs/content/docs/api/endpoints.mdx

@@ -634,6 +634,8 @@ Cloud-managed OAuth additionally exposes `POST /pool-link/instance/oauth/start-c
 
 `POST /internal/worker/warmup-dispatch` is an execution-plane internal-token route, not a public JWT/API-key scope. It inspects, begins and durably finishes a task/mailbox/worker/nonce-bound command. Start repeats authoritative policy checks; unknown or denied authority cannot permit native execution. Receipt/header/model text supplies no authorization.
 
+`POST /internal/email-message-map/arrival` uses the existing internal node token. Only `204` with `X-Warmbly-Arrival-Durable: 1` acknowledges durable admission. Authoritative absence of the exact mailbox/user ownership returns `503` with `X-Warmbly-Arrival-Ownership-Lost` containing the requested mailbox UUID and no durable acknowledgment. Updated workers retire only that mailbox generation; the failed arrival remains unacknowledged, its stored body is retained and its provider cursor is not advanced. Ordinary `503` responses, transport failures, database failures and encryption failures remain retryable. Older workers ignore the new marker and continue holding the arrival; an upgraded backend and worker are both required for retirement. This does not clear send holds, resend messages or alter Kafka offsets.
+
 `POST /internal/worker/outbound-dispatch`, `/internal/worker/warmup-actions`, and `/internal/worker/diagnostic-auth` use the existing execution-plane internal token, never customer API keys. They bind the current mailbox/worker and require protocol 2. Outbound start additionally rechecks task/organization/provider/nonce/recipients and current campaign or diagnostic policy. Diagnostic authorization/result submission binds one exact already-owned warmup receipt and short-lived nonce; it neither ingests unrelated inbox messages nor trusts event/header claims. No public scope or permission was added. Details and limits: [send safety](/development/send-safety/).
 
 ### Referrals and discounts

internal/api/handler/internal_arrival_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
@@ -25,6 +26,33 @@ type internalArrivalMap struct {
 	err                error
 }
 
+func TestInternalArrivalOwnershipLossIsNotDurableAdmission(t *testing.T) {
+	mailbox := uuid.NewString()
+	for _, tc := range []struct {
+		name, mailbox, marker string
+		err                   error
+	}{
+		{"ownership-loss", mailbox, mailbox, fmt.Errorf("wrapped: %w", repository.ErrArrivalMailboxOwnershipLost)},
+		{"database-failure", mailbox, "", &pgconn.PgError{Code: "08006", Message: "private payload"}},
+		{"unrelated-missing-row", mailbox, "", pgx.ErrNoRows},
+		{"unconfirmed", mailbox, "", repository.ErrArrivalAdmissionUnconfirmed},
+		{"invalid-id", "invalid", "", repository.ErrArrivalMailboxOwnershipLost},
+		{"nil-id", uuid.Nil.String(), "", repository.ErrArrivalMailboxOwnershipLost},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := &internalArrivalMap{err: tc.err}
+			h := &Handler{EmailMessageMap: m}
+			r := gin.New()
+			r.POST("/arrival", h.InternalAdmitEmailArrival)
+			w := httptest.NewRecorder()
+			r.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/arrival", strings.NewReader(fmt.Sprintf(`{"map":{"email_id":%q},"pending":{}}`, tc.mailbox))))
+			if m.calls != 1 || w.Code != 503 || w.Body.Len() != 0 || w.Header().Get("X-Warmbly-Arrival-Durable") != "" || w.Header().Get("X-Warmbly-Arrival-Ownership-Lost") != tc.marker {
+				t.Fatal("failure acknowledged or ownership marker inferred without authority")
+			}
+		})
+	}
+}
+
 func TestInternalArrivalLogsOnlySafeDiagnosticsWithoutAcknowledging(t *testing.T) {
 	var output bytes.Buffer
 	original := log.Logger

internal/api/handler/internal_email_message_map.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 
@@ -30,6 +31,11 @@ func (h *Handler) InternalAdmitEmailArrival(c *gin.Context) {
 	}
 	if err := repo.AdmitArrival(c.Request.Context(), repository.EmailMessageData(p.Map), p.Pending); err != nil {
 		logArrivalAdmissionFailure(c, p.Map, err)
+		if errors.Is(err, repository.ErrArrivalMailboxOwnershipLost) {
+			if mailbox, parseErr := uuid.Parse(p.Map.EmailID); parseErr == nil && mailbox != uuid.Nil {
+				c.Header("X-Warmbly-Arrival-Ownership-Lost", mailbox.String())
+			}
+		}
 		// Do not expose message content or org encryption failures over the protocol.
 		c.Status(http.StatusServiceUnavailable)
 		return

internal/app/worker/wmail/admit.go

@@ -203,6 +203,11 @@ func (w *WMail) storeNew(ctx context.Context, msg *models.EmailMessageData, data
 	}
 	if durable, ok := w.EmailMessageMapRepository.(repository.ArrivalAdmission); ok {
 		err := durable.AdmitArrival(ctx, mapping, &repository.PendingArrival{Arrival: arrival, Bounce: bounce, Complaint: complaint})
+		if errors.Is(err, repository.ErrArrivalMailboxOwnershipLost) {
+			log.Warn().Str("email_id", w.ID.String()).Msg("sync: mailbox ownership revoked; worker generation retired, arrival unacknowledged")
+			w.Terminate()
+			return err
+		}
 		if !errors.Is(err, repository.ErrArrivalOutboxUnsupported) {
 			return err
 		}

internal/app/worker/wmail/arrival_durability_test.go

@@ -3,9 +3,17 @@ package wmail
 import (
 	"context"
 	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
+	"time"
 
+	goimap "github.com/emersion/go-imap/v2"
 	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/client/msgraph"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 )
@@ -15,9 +23,13 @@ type durableTestMap struct {
 	pending         map[string]*repository.PendingArrival
 	failAfterCommit bool
 	unsupported     bool
+	admitErr        error
 }
 
 func (m *durableTestMap) AdmitArrival(ctx context.Context, data repository.EmailMessageData, p *repository.PendingArrival) error {
+	if m.admitErr != nil {
+		return m.admitErr
+	}
 	if m.unsupported {
 		return repository.ErrArrivalOutboxUnsupported
 	}
@@ -31,6 +43,174 @@ func (m *durableTestMap) AdmitArrival(ctx context.Context, data repository.Email
 	return nil
 }
 
+type retainedArrivalStore struct {
+	fakeStore
+	body    map[string][]byte
+	deletes int
+}
+
+func (s *retainedArrivalStore) Put(_ context.Context, key string, body io.Reader, _ string) error {
+	data, err := io.ReadAll(body)
+	if err == nil {
+		s.body[key] = data
+	}
+	return err
+}
+
+func (s *retainedArrivalStore) Delete(context.Context, string) error {
+	s.deletes++
+	return nil
+}
+
+func TestArrivalOwnershipLossRetiresOnlyExactGenerationAcrossProviders(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		provider models.InboxProvider
+		store    func(*WMail, context.Context, *models.EmailMessageData) error
+	}{
+		{"google", models.InboxProviderGoogle, (*WMail).googleStore},
+		{"graph", models.InboxProviderOutlook, (*WMail).graphStore},
+		{"imap", models.InboxProviderSMTPIMAP, (*WMail).imapStore},
+	} {
+		for _, loss := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/ownershipLost=%t", tc.name, loss), func(t *testing.T) {
+				failure := errors.New("arrival admission not confirmed: status 503")
+				if loss {
+					failure = fmt.Errorf("wrapped: %w", repository.ErrArrivalMailboxOwnershipLost)
+				}
+				m := &durableTestMap{recoveryMessageMap: recoveryMessageMap{data: map[string]repository.EmailMessageData{}}, pending: map[string]*repository.PendingArrival{}, admitErr: failure}
+				var events []captured
+				w := newTestWMail(t, tc.provider, &events)
+				w.EmailMessageMapRepository = m
+				storage := &retainedArrivalStore{body: map[string][]byte{}}
+				w.Storage = storage
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				w.Ctx, w.Cancel = ctx, cancel
+				terminations := 0
+				w.TerminateFunc = func() { terminations++ }
+				msg := &models.EmailMessageData{MessageID: "<held@test.local>", GmailID: "held-provider-id", BodyPlain: "retained body"}
+				if err := tc.store(w, ctx, msg); !errors.Is(err, failure) {
+					t.Fatalf("failed arrival was acknowledged: %v", err)
+				}
+				if len(storage.body) != 1 || storage.deletes != 0 || len(m.data) != 0 || len(m.pending) != 0 || len(events) != 0 {
+					t.Fatal("failed admission discarded body, created mapping or published success")
+				}
+				if (terminations == 1) != loss || (ctx.Err() != nil) != loss {
+					t.Fatal("ordinary failure retired mailbox or authoritative absence did not")
+				}
+				done, allowed := w.BeginExecution()
+				if allowed == loss {
+					t.Fatal("retired mailbox admitted new execution or retryable mailbox stopped")
+				}
+				if allowed {
+					done()
+				}
+			})
+		}
+	}
+}
+
+func TestArrivalOwnershipLossHoldsIMAPCursorAndSyncTruth(t *testing.T) {
+	for _, loss := range []bool{false, true} {
+		t.Run(fmt.Sprintf("ownershipLost=%t", loss), func(t *testing.T) {
+			failure := errors.New("arrival admission not confirmed: status 503")
+			if loss {
+				failure = repository.ErrArrivalMailboxOwnershipLost
+			}
+			box := &models.Mailbox{Name: "INBOX", UIDValidity: 1, UIDNext: 1, HighestModSeq: 1}
+			conn := &fakeImapConn{folders: []models.Mailbox{{Name: "INBOX", UIDValidity: 1, UIDNext: 2, HighestModSeq: 2}}, changed: []goimap.UID{1}}
+			w, events := newIMAPTestMail(conn, &fixedBudget{allow: 10}, box)
+			w.EmailMessageMapRepository = &durableTestMap{recoveryMessageMap: recoveryMessageMap{data: map[string]repository.EmailMessageData{}}, pending: map[string]*repository.PendingArrival{}, admitErr: failure}
+			storage := &retainedArrivalStore{body: map[string][]byte{}}
+			w.Storage = storage
+			last := time.Now().Add(-time.Hour)
+			w.tracker.state.LastSyncedAt = &last
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			w.Ctx, w.Cancel = ctx, cancel
+			terminated := false
+			w.TerminateFunc = func() { terminated = true }
+			if err := w.Sync(ctx); err != nil {
+				t.Fatal(err)
+			}
+			if box.UIDNext != 1 || box.HighestModSeq != 1 || !w.tracker.state.LastSyncedAt.Equal(last) || w.tracker.tickComplete || len(newEmails(*events)) != 0 {
+				t.Fatal("failed admission advanced cursor, sync success or arrival acknowledgment")
+			}
+			if len(storage.body) != 1 || storage.deletes != 0 || terminated != loss {
+				t.Fatal("incorrect body retention or mailbox retirement")
+			}
+		})
+	}
+}
+
+func TestArrivalOwnershipLossHoldsOAuthProviderCheckpoints(t *testing.T) {
+	for _, provider := range []string{"google", "graph"} {
+		for _, loss := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/ownershipLost=%t", provider, loss), func(t *testing.T) {
+				var events []captured
+				var relayed []models.SyncState
+				var w *WMail
+				if provider == "google" {
+					srv := httptest.NewServer(http.HandlerFunc(func(out http.ResponseWriter, r *http.Request) {
+						out.Header().Set("Content-Type", "application/json")
+						if strings.HasSuffix(r.URL.Path, "/history") {
+							_, _ = out.Write([]byte(`{"history":[{"id":"101","messagesAdded":[{"message":{"id":"new","threadId":"thread"}}]}],"historyId":"200"}`))
+							return
+						}
+						_, _ = out.Write([]byte(`{"id":"new","threadId":"thread","labelIds":["INBOX"],"payload":{"headers":[{"name":"Message-ID","value":"<new@test.local>"}]}}`))
+					}))
+					defer srv.Close()
+					w = newGoogleTestMail(t, srv, &events)
+					w.GoogleData.LastHistoryID = 100
+				} else {
+					g := newFakeGraph()
+					g.live = []string{"new"}
+					w = newGraphTestMail(t, g.serve(t), &events, &relayed)
+				}
+				previous := seedSyncTruth(w, &relayed)
+				failure := errors.New("arrival admission not confirmed: status 503")
+				if loss {
+					failure = repository.ErrArrivalMailboxOwnershipLost
+				}
+				w.EmailMessageMapRepository = &durableTestMap{recoveryMessageMap: recoveryMessageMap{data: map[string]repository.EmailMessageData{}}, pending: map[string]*repository.PendingArrival{}, admitErr: failure}
+				storage := &retainedArrivalStore{body: map[string][]byte{}}
+				w.Storage = storage
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				w.Ctx, w.Cancel = ctx, cancel
+				terminated := false
+				w.TerminateFunc = func() { terminated = true }
+				if provider == "google" {
+					if err := w.SyncGoogle(ctx); err != nil {
+						t.Fatal(err)
+					}
+					if w.GoogleData.LastHistoryID != 100 {
+						t.Fatal("failed admission advanced Gmail history")
+					}
+				} else {
+					before := w.GraphData.Client.DeltaLinks[msgraph.FolderInbox]
+					if err := w.SyncGraph(ctx); err != nil {
+						t.Fatal(err)
+					}
+					if w.GraphData.Client.DeltaLinks[msgraph.FolderInbox] != before {
+						t.Fatal("failed admission advanced Graph delta")
+					}
+				}
+				assertNoSyncSuccess(t, w, previous)
+				if len(storage.body) != 1 || storage.deletes != 0 || terminated != loss || w.tracker.tickComplete {
+					t.Fatal("incorrect retention, retirement or sync success")
+				}
+				for _, event := range events {
+					if event.eventType == models.JobEventTypeNewEmail || event.eventType == models.JobEventTypeHistoryIDUpdate || event.eventType == models.JobEventTypeGraphDeltaUpdate {
+						t.Fatal("failed admission published arrival or advanced checkpoint")
+					}
+				}
+			})
+		}
+	}
+}
+
 func TestArrivalDurabilityRestartPreservesIDAcrossProviders(t *testing.T) {
 	for _, tc := range []struct {
 		name     string

internal/repository/arrival_admission_error_test.go

@@ -1,11 +1,13 @@
 package repository
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
 	"testing"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 )
 
@@ -30,6 +32,23 @@ func TestArrivalAdmissionDiagnosticNeverExposesErrorText(t *testing.T) {
 	}
 }
 
+func TestArrivalOwnershipLossRequiresAuthoritativeAbsence(t *testing.T) {
+	for _, stage := range []string{"mailbox_ownership", "mailbox_ownership_recheck"} {
+		for _, cause := range []error{pgx.ErrNoRows, fmt.Errorf("wrapped: %w", pgx.ErrNoRows), context.Canceled, context.DeadlineExceeded, errors.New("database unavailable"), &pgconn.PgError{Code: "08006"}} {
+			err := arrivalOwnershipFailure(stage, cause)
+			if errors.Is(err, ErrArrivalMailboxOwnershipLost) != errors.Is(cause, pgx.ErrNoRows) || !errors.Is(err, cause) {
+				t.Fatalf("ownership error incorrectly classified: %v", cause)
+			}
+			if got, _ := ArrivalAdmissionDiagnostic(err); got != stage {
+				t.Fatalf("lost stage: %s", got)
+			}
+		}
+	}
+	if errors.Is(arrivalFailure("organization_key", pgx.ErrNoRows), ErrArrivalMailboxOwnershipLost) {
+		t.Fatal("unrelated missing row retired mailbox")
+	}
+}
+
 func TestArrivalAdmissionFailurePreservesCauseAndHidesText(t *testing.T) {
 	cause := errors.New("private payload")
 	err := arrivalFailure("payload_encoding", cause)

internal/repository/email_message_map.go

@@ -25,6 +25,7 @@ type EmailMessageMapRepository interface {
 
 var ErrArrivalOutboxUnsupported = errors.New("durable arrival admission unavailable; upgrade backend and consumer first")
 var ErrArrivalAdmissionUnconfirmed = errors.New("existing provider mapping has no verifiable pending admission")
+var ErrArrivalMailboxOwnershipLost = errors.New("arrival mailbox is no longer owned by this user")
 
 // ArrivalAdmission is an optional internal protocol, separate from legacy map writes.
 type ArrivalAdmission interface {

internal/repository/http_email_message_map.go

@@ -115,6 +115,13 @@ func (r *httpEmailMessageMapRepository) AdmitArrival(ctx context.Context, data E
 	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusMethodNotAllowed {
 		return ErrArrivalOutboxUnsupported
 	}
+	if resp.StatusCode == http.StatusServiceUnavailable && resp.Header.Get("X-Warmbly-Arrival-Durable") == "" &&
+		resp.Request != nil && resp.Request.URL.String() == req.URL.String() && resp.Request.Method == req.Method {
+		markers := resp.Header.Values("X-Warmbly-Arrival-Ownership-Lost")
+		if mailbox, parseErr := uuid.Parse(data.EmailID); parseErr == nil && mailbox != uuid.Nil && len(markers) == 1 && markers[0] == mailbox.String() {
+			return ErrArrivalMailboxOwnershipLost
+		}
+	}
 	if resp.StatusCode != http.StatusNoContent || resp.Header.Get("X-Warmbly-Arrival-Durable") != "1" {
 		return fmt.Errorf("arrival admission not confirmed: status %d", resp.StatusCode)
 	}

internal/repository/http_sync_arrival_test.go

@@ -7,6 +7,8 @@ import (
 	"net/http/httptest"
 	"sync/atomic"
 	"testing"
+
+	"github.com/google/uuid"
 )
 
 func TestHTTPArrivalAdmissionUpgradeAcknowledgement(t *testing.T) {
@@ -41,6 +43,50 @@ func TestHTTPArrivalAdmissionUpgradeAcknowledgement(t *testing.T) {
 	}
 }
 
+func TestHTTPArrivalOwnershipLossRequiresExactNonAcknowledgedMarker(t *testing.T) {
+	mailbox := uuid.NewString()
+	for _, tc := range []struct {
+		name, requested, ack string
+		status               int
+		markers              []string
+		lost                 bool
+	}{
+		{"matching", mailbox, "", 503, []string{mailbox}, true},
+		{"missing", mailbox, "", 503, nil, false},
+		{"other-mailbox", mailbox, "", 503, []string{uuid.NewString()}, false},
+		{"invalid", mailbox, "", 503, []string{"private error"}, false},
+		{"duplicate", mailbox, "", 503, []string{mailbox, mailbox}, false},
+		{"invalid-request", "invalid", "", 503, []string{"invalid"}, false},
+		{"nil-mailbox", uuid.Nil.String(), "", 503, []string{uuid.Nil.String()}, false},
+		{"wrong-status", mailbox, "", 500, []string{mailbox}, false},
+		{"auth-failure", mailbox, "", 401, []string{mailbox}, false},
+		{"contradictory-durable", mailbox, "1", 503, []string{mailbox}, false},
+		{"not-durable-success", mailbox, "", 204, []string{mailbox}, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.Header.Get("Authorization") != "Bearer fixture-token" {
+					t.Error("request was not authenticated")
+				}
+				for _, marker := range tc.markers {
+					w.Header().Add("X-Warmbly-Arrival-Ownership-Lost", marker)
+				}
+				w.Header().Set("X-Warmbly-Arrival-Durable", tc.ack)
+				w.WriteHeader(tc.status)
+			}))
+			defer srv.Close()
+			repo, err := NewHTTPEmailMessageMapRepository(srv.URL, "fixture-token")
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = repo.(ArrivalAdmission).AdmitArrival(t.Context(), EmailMessageData{EmailID: tc.requested}, &PendingArrival{})
+			if err == nil || errors.Is(err, ErrArrivalMailboxOwnershipLost) != tc.lost || errors.Is(err, ErrArrivalOutboxUnsupported) {
+				t.Fatalf("unexpected admission: %v", err)
+			}
+		})
+	}
+}
+
 func TestHTTPArrivalAdmissionResponseLossDoesNotDowngrade(t *testing.T) {
 	var committed atomic.Bool
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -65,3 +111,24 @@ func TestHTTPArrivalAdmissionResponseLossDoesNotDowngrade(t *testing.T) {
 		t.Fatalf("response loss treated as unsupported: committed=%v err=%v", committed.Load(), err)
 	}
 }
+
+func TestHTTPArrivalRedirectCannotAuthorizeMailboxRetirement(t *testing.T) {
+	mailbox := uuid.NewString()
+	redirected := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("X-Warmbly-Arrival-Ownership-Lost", mailbox)
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer redirected.Close()
+	original := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Redirect(w, r, redirected.URL, http.StatusTemporaryRedirect)
+	}))
+	defer original.Close()
+	repo, err := NewHTTPEmailMessageMapRepository(original.URL, "fixture-token")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = repo.(ArrivalAdmission).AdmitArrival(t.Context(), EmailMessageData{EmailID: mailbox}, &PendingArrival{})
+	if err == nil || errors.Is(err, ErrArrivalMailboxOwnershipLost) || errors.Is(err, ErrArrivalOutboxUnsupported) {
+		t.Fatalf("redirected endpoint authorized retirement: %v", err)
+	}
+}

internal/repository/pg_sync_arrival.go

@@ -47,7 +47,7 @@ func (r *pgEmailMessageMapRepository) AdmitArrival(ctx context.Context, data Ema
 	var org uuid.UUID
 	err = r.db.QueryRow(ctx, `SELECT organization_id FROM email_accounts WHERE id=$1 AND user_id=$2`, email, user).Scan(&org)
 	if err != nil {
-		return arrivalFailure("mailbox_ownership", err)
+		return arrivalOwnershipFailure("mailbox_ownership", err)
 	}
 	sealed, err := r.cipher.Cipher(ctx, org)
 	if err != nil {
@@ -69,7 +69,7 @@ func (r *pgEmailMessageMapRepository) AdmitArrival(ctx context.Context, data Ema
 	// Recheck and lock ownership after crypto, without nesting key-store I/O in the transaction.
 	var owned uuid.UUID
 	if err = tx.QueryRow(ctx, `SELECT id FROM email_accounts WHERE id=$1 AND user_id=$2 AND organization_id=$3 FOR SHARE`, email, user, org).Scan(&owned); err != nil {
-		return arrivalFailure("mailbox_ownership_recheck", err)
+		return arrivalOwnershipFailure("mailbox_ownership_recheck", err)
 	}
 	// A retry after an ambiguous response adopts the first canonical identity.
 	inserted, err := tx.Exec(ctx, `INSERT INTO email_message_map(user_id,email_id,message_id,id,thread_id)
@@ -102,6 +102,13 @@ func (r *pgEmailMessageMapRepository) AdmitArrival(ctx context.Context, data Ema
 	return nil
 }
 
+func arrivalOwnershipFailure(stage string, err error) error {
+	if errors.Is(err, pgx.ErrNoRows) {
+		err = errors.Join(ErrArrivalMailboxOwnershipLost, err)
+	}
+	return arrivalFailure(stage, err)
+}
+
 func (r *pgEmailMessageMapRepository) ArrivalBacklog(ctx context.Context) (ArrivalBacklog, error) {
 	var backlog ArrivalBacklog
 	err := r.db.QueryRow(ctx, `SELECT count(*),min(created_at) FROM sync_arrival_outbox`).Scan(&backlog.Pending, &backlog.Oldest)

internal/repository/pg_sync_arrival_live_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/warmbly/warmbly/internal/app/cipher"
 	"github.com/warmbly/warmbly/internal/infrastructure/db"
@@ -25,6 +26,52 @@ func arrivalLiveCipher(t *testing.T, d *db.DB) cipher.CipherService {
 	return cipher.NewService(k, nil, encryptedkeys.NewPostgres(d))
 }
 
+type arrivalRecheckCipher struct {
+	cipher.CipherService
+	after func()
+	err   error
+}
+
+func (c arrivalRecheckCipher) Cipher(ctx context.Context, org uuid.UUID) (*cipher.Cipher, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	sealed, err := c.CipherService.Cipher(ctx, org)
+	if err == nil && c.after != nil {
+		c.after()
+	}
+	return sealed, err
+}
+
+func TestLiveSyncArrivalOwnershipRecheckRetiresOnlyAuthoritativeAbsence(t *testing.T) {
+	d := liveUniboxFolderDB(t)
+	f := newUniboxFolderFixture(t, d.Pool)
+	data, pending := arrivalLivePayload(f, "ownership-recheck")
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	r := NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d))
+	if err := r.AdmitArrival(ctx, data, pending); !errors.Is(err, context.Canceled) || errors.Is(err, ErrArrivalMailboxOwnershipLost) {
+		t.Fatalf("canceled ownership lookup retired mailbox: %v", err)
+	}
+	r = NewDurableEmailMessageMapRepository(d, arrivalRecheckCipher{err: pgx.ErrNoRows})
+	if err := r.AdmitArrival(t.Context(), data, pending); !errors.Is(err, pgx.ErrNoRows) || errors.Is(err, ErrArrivalMailboxOwnershipLost) {
+		t.Fatalf("missing organization key retired mailbox: %v", err)
+	}
+	r = NewDurableEmailMessageMapRepository(d, arrivalRecheckCipher{CipherService: arrivalLiveCipher(t, d), after: func() {
+		if _, err := d.Exec(t.Context(), `DELETE FROM email_accounts WHERE id=$1`, f.mailbox); err != nil {
+			t.Fatal(err)
+		}
+	}})
+	err := r.AdmitArrival(t.Context(), data, pending)
+	stage, _ := ArrivalAdmissionDiagnostic(err)
+	if !errors.Is(err, ErrArrivalMailboxOwnershipLost) || !errors.Is(err, pgx.ErrNoRows) || stage != "mailbox_ownership_recheck" {
+		t.Fatalf("ownership recheck did not retire removed mailbox: %v stage=%s", err, stage)
+	}
+	if known, err := r.Get(t.Context(), f.user, f.mailbox, data.MessageID); err != nil || known != nil {
+		t.Fatal("failed ownership recheck created a mapping")
+	}
+}
+
 func arrivalLivePayload(f *uniboxFolderFixture, key string) (EmailMessageData, *PendingArrival) {
 	id := uuid.New()
 	now := time.Now().UTC()
@@ -217,7 +264,7 @@ func TestLiveSyncArrivalReportStagesOwnershipAndLegacyMaps(t *testing.T) {
 	p.Arrival.UserID = wrong
 	p.Bounce.UserID = wrong
 	p.Complaint.UserID = wrong
-	if err := r.AdmitArrival(ctx, data, p); err == nil {
+	if err := r.AdmitArrival(ctx, data, p); !errors.Is(err, ErrArrivalMailboxOwnershipLost) || !errors.Is(err, pgx.ErrNoRows) {
 		t.Fatal("another user admitted mail")
 	}
 	data.UserID = f.user.String()
@@ -292,6 +339,8 @@ func TestLiveSyncArrivalAtomicRollbackDeletedMailboxAndLease(t *testing.T) {
 	})
 	if err := r.AdmitArrival(ctx, data, p); err == nil {
 		t.Fatal("expected outbox insertion failure")
+	} else if errors.Is(err, ErrArrivalMailboxOwnershipLost) {
+		t.Fatal("database failure retired an owned mailbox")
 	}
 	if known, err := r.Get(ctx, f.user, f.mailbox, data.MessageID); err != nil || known != nil {
 		t.Fatal("map escaped rolled-back admission")
@@ -340,6 +389,9 @@ func TestLiveSyncArrivalAtomicRollbackDeletedMailboxAndLease(t *testing.T) {
 	if _, err := d.Exec(ctx, `DELETE FROM email_accounts WHERE id=$1`, f.mailbox); err != nil {
 		t.Fatal(err)
 	}
+	if err := r.AdmitArrival(ctx, data, p); !errors.Is(err, ErrArrivalMailboxOwnershipLost) || !errors.Is(err, pgx.ErrNoRows) {
+		t.Fatalf("deleted mailbox did not fail authoritative ownership: %v", err)
+	}
 	if pending, err := r.HasPendingArrival(ctx, f.user, f.mailbox, p.Arrival.Message.ID); err != nil || pending {
 		t.Fatal("deleted mailbox retained pending arrival")
 	}