From f2876f1fc9db06d2762610f0d0dcbaf1f00738c8 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 08:17:59 +0000 Subject: [PATCH] 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 | 2 + internal/api/handler/internal_arrival_test.go | 28 +++ .../api/handler/internal_email_message_map.go | 6 + internal/app/worker/wmail/admit.go | 5 + .../worker/wmail/arrival_durability_test.go | 180 ++++++++++++++++++ .../arrival_admission_error_test.go | 19 ++ internal/repository/email_message_map.go | 1 + internal/repository/http_email_message_map.go | 7 + internal/repository/http_sync_arrival_test.go | 67 +++++++ internal/repository/pg_sync_arrival.go | 11 +- .../repository/pg_sync_arrival_live_test.go | 54 +++++- 11 files changed, 377 insertions(+), 3 deletions(-) diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index f765c3ac3..f541ca8a5 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/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 diff --git a/internal/api/handler/internal_arrival_test.go b/internal/api/handler/internal_arrival_test.go index bd4a591cc..c5e1e5513 100644 --- a/internal/api/handler/internal_arrival_test.go +++ b/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 diff --git a/internal/api/handler/internal_email_message_map.go b/internal/api/handler/internal_email_message_map.go index 7853b1c71..ffefd77d1 100644 --- a/internal/api/handler/internal_email_message_map.go +++ b/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 diff --git a/internal/app/worker/wmail/admit.go b/internal/app/worker/wmail/admit.go index e6dba5690..4e9f8ddeb 100644 --- a/internal/app/worker/wmail/admit.go +++ b/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 } diff --git a/internal/app/worker/wmail/arrival_durability_test.go b/internal/app/worker/wmail/arrival_durability_test.go index e2e5bf935..764f55363 100644 --- a/internal/app/worker/wmail/arrival_durability_test.go +++ b/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: "", 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":""}]}}`)) + })) + 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 diff --git a/internal/repository/arrival_admission_error_test.go b/internal/repository/arrival_admission_error_test.go index 809785d88..9cc1d88ac 100644 --- a/internal/repository/arrival_admission_error_test.go +++ b/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) diff --git a/internal/repository/email_message_map.go b/internal/repository/email_message_map.go index f2c4b530d..e61413e80 100644 --- a/internal/repository/email_message_map.go +++ b/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 { diff --git a/internal/repository/http_email_message_map.go b/internal/repository/http_email_message_map.go index 0c9e6f2d1..e1f35e132 100644 --- a/internal/repository/http_email_message_map.go +++ b/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) } diff --git a/internal/repository/http_sync_arrival_test.go b/internal/repository/http_sync_arrival_test.go index 4853fcae5..084e6a89f 100644 --- a/internal/repository/http_sync_arrival_test.go +++ b/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) + } +} diff --git a/internal/repository/pg_sync_arrival.go b/internal/repository/pg_sync_arrival.go index 9101fdb8e..831ba6c41 100644 --- a/internal/repository/pg_sync_arrival.go +++ b/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) diff --git a/internal/repository/pg_sync_arrival_live_test.go b/internal/repository/pg_sync_arrival_live_test.go index f09754d0e..be2766396 100644 --- a/internal/repository/pg_sync_arrival_live_test.go +++ b/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") }