From 4f0cd134c0829c8de16ee60e71e0ab883a23cbf7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 12:17:17 +0000 Subject: [PATCH] feat: atomically retain encrypted mailbox arrivals for original-ID consumer replay and retry Kafka handlers before advancing offsets --- cmd/backend/main.go | 1 + cmd/consumer/main.go | 1 + internal/api/handler/internal_arrival_test.go | 69 +++++ .../api/handler/internal_email_message_map.go | 23 ++ internal/api/routes.go | 1 + internal/app/consumer/arrival_outbox.go | 40 +++ .../app/consumer/arrival_outbox_live_test.go | 151 ++++++++++ internal/app/consumer/arrival_outbox_test.go | 21 ++ internal/app/consumer/service.go | 2 + internal/app/orgtransfer/spec.go | 4 +- .../sync_arrival_scope_live_test.go | 63 ++++ internal/app/worker/wmail/admit.go | 20 +- .../worker/wmail/arrival_durability_test.go | 97 ++++++ .../000278_sync_arrival_outbox.down.sql | 8 + .../000278_sync_arrival_outbox.up.sql | 18 ++ .../db/sync_arrival_upgrade_test.go | 106 +++++++ internal/infrastructure/kafka/consumer.go | 36 ++- .../kafka/consumer_retry_test.go | 85 ++++++ internal/repository/email_message_map.go | 20 ++ internal/repository/http_email_message_map.go | 28 ++ internal/repository/http_sync_arrival_test.go | 67 +++++ internal/repository/pg_email_message_map.go | 18 +- internal/repository/pg_sync_arrival.go | 176 +++++++++++ .../repository/pg_sync_arrival_live_test.go | 276 ++++++++++++++++++ 24 files changed, 1315 insertions(+), 16 deletions(-) create mode 100644 internal/api/handler/internal_arrival_test.go create mode 100644 internal/app/consumer/arrival_outbox.go create mode 100644 internal/app/consumer/arrival_outbox_live_test.go create mode 100644 internal/app/consumer/arrival_outbox_test.go create mode 100644 internal/app/orgtransfer/sync_arrival_scope_live_test.go create mode 100644 internal/app/worker/wmail/arrival_durability_test.go create mode 100644 internal/infrastructure/db/migrations/000278_sync_arrival_outbox.down.sql create mode 100644 internal/infrastructure/db/migrations/000278_sync_arrival_outbox.up.sql create mode 100644 internal/infrastructure/db/sync_arrival_upgrade_test.go create mode 100644 internal/infrastructure/kafka/consumer_retry_test.go create mode 100644 internal/repository/http_sync_arrival_test.go create mode 100644 internal/repository/pg_sync_arrival.go create mode 100644 internal/repository/pg_sync_arrival_live_test.go diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 1d2497af7..a1d9e5885 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -1137,6 +1137,7 @@ func main() { } cipherService = cipher.NewService(kms, cache, encryptedKeys) + emailMessageMapForHandler = repository.NewDurableEmailMessageMapRepository(primaryDB, cipherService) // Third-party integrations: OAuth connect flows + encrypted token // storage (sealed with the connecting user's DEK) + event-driven actions. diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 2efa4372b..d00d402ca 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -526,6 +526,7 @@ func main() { EmailHistoryIDRepository: emailHistoryIDRepo, EmailGraphDeltaRepository: emailGraphDeltaRepo, EmailSyncStateRepository: repository.NewEmailSyncStateRepository(primaryDB), + ArrivalOutbox: repository.NewDurableEmailMessageMapRepository(primaryDB, cipherService), EmailAccountErrorRepository: emailAccountErrorRepo, WarmupRepo: warmupRepo, PoolLinkRepo: repository.NewPoolLinkRepository(primaryDB.Pool), diff --git a/internal/api/handler/internal_arrival_test.go b/internal/api/handler/internal_arrival_test.go new file mode 100644 index 000000000..fc5e70490 --- /dev/null +++ b/internal/api/handler/internal_arrival_test.go @@ -0,0 +1,69 @@ +package handler + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/repository" +) + +type internalArrivalMap struct { + calls, legacyCalls int + err error +} + +func (m *internalArrivalMap) Add(context.Context, repository.EmailMessageData) error { + m.legacyCalls++ + return nil +} +func (*internalArrivalMap) Get(context.Context, uuid.UUID, uuid.UUID, string) (*repository.EmailMessageData, error) { + return nil, nil +} +func (*internalArrivalMap) Del(context.Context, uuid.UUID, uuid.UUID, string, uuid.UUID) error { + return nil +} +func (m *internalArrivalMap) AdmitArrival(context.Context, repository.EmailMessageData, *repository.PendingArrival) error { + m.calls++ + return m.err +} + +func TestInternalArrivalAcknowledgesOnlyCompletedAdmission(t *testing.T) { + for _, tc := range []struct { + name string + err error + status int + ack string + }{ + {"success", nil, 204, "1"}, {"failure", errors.New("private payload or cipher failure"), 503, ""}, + } { + 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(`{"map":{},"pending":{}}`))) + if m.calls != 1 || w.Code != tc.status || w.Header().Get("X-Warmbly-Arrival-Durable") != tc.ack || w.Body.Len() != 0 { + t.Fatalf("calls=%d status=%d body=%q", m.calls, w.Code, w.Body.String()) + } + }) + } +} + +func TestInternalArrivalPreservesLegacyWorkerProtocol(t *testing.T) { + m := &internalArrivalMap{} + h := &Handler{EmailMessageMap: m} + r := gin.New() + r.PUT("/map", h.InternalPutEmailMessageMap) + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodPut, "/map", strings.NewReader(`{"user_id":"fixture","email_id":"fixture","message_id":"fixture","id":"fixture","thread_id":"fixture"}`))) + if w.Code != 204 || m.legacyCalls != 1 || m.calls != 0 || w.Header().Get("X-Warmbly-Arrival-Durable") != "" { + t.Fatalf("legacy contract changed: %d", w.Code) + } +} diff --git a/internal/api/handler/internal_email_message_map.go b/internal/api/handler/internal_email_message_map.go index ca320d69f..b6fc83659 100644 --- a/internal/api/handler/internal_email_message_map.go +++ b/internal/api/handler/internal_email_message_map.go @@ -11,6 +11,29 @@ import ( "github.com/warmbly/warmbly/internal/repository" ) +func (h *Handler) InternalAdmitEmailArrival(c *gin.Context) { + var p struct { + Map emailMessageMapPayload `json:"map"` + Pending *repository.PendingArrival `json:"pending"` + } + if err := json.NewDecoder(http.MaxBytesReader(c.Writer, c.Request.Body, 4<<20)).Decode(&p); err != nil { + c.Status(http.StatusBadRequest) + return + } + repo, ok := h.EmailMessageMap.(repository.ArrivalAdmission) + if !ok { + c.Status(http.StatusServiceUnavailable) + return + } + if err := repo.AdmitArrival(c.Request.Context(), repository.EmailMessageData(p.Map), p.Pending); err != nil { + // Do not expose message content or org encryption failures over the protocol. + c.Status(http.StatusServiceUnavailable) + return + } + c.Header("X-Warmbly-Arrival-Durable", "1") + c.Status(http.StatusNoContent) +} + // Internal email-message-map endpoints. Workers call these instead of touching // Postgres directly (per CLAUDE.md), mirroring the DEK endpoints. Auth via // middleware.NodeBrokerAuthMiddleware (the node token). diff --git a/internal/api/routes.go b/internal/api/routes.go index d68d5feee..45fb66c7f 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -191,6 +191,7 @@ func Run( // Worker mailbox-sync messageId -> internal email map. node.GET("/email-message-map", h.InternalGetEmailMessageMap) node.PUT("/email-message-map", h.InternalPutEmailMessageMap) + node.POST("/email-message-map/arrival", h.InternalAdmitEmailArrival) node.DELETE("/email-message-map", h.InternalDeleteEmailMessageMap) // Sync governor priority lane: "is this new message a reply to diff --git a/internal/app/consumer/arrival_outbox.go b/internal/app/consumer/arrival_outbox.go new file mode 100644 index 000000000..6df1207ee --- /dev/null +++ b/internal/app/consumer/arrival_outbox.go @@ -0,0 +1,40 @@ +package jobs + +import ( + "context" + "fmt" + "time" + + "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/models" +) + +func (s *JobsService) runArrivalOutbox(ctx context.Context) { + if s.ArrivalOutbox == nil { + return + } + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + pass, cancel := context.WithTimeout(ctx, 30*time.Second) + err := s.ArrivalOutbox.DeliverArrivals(pass, func(ctx context.Context, kind models.JobEventType, body any) error { + return s.deliverArrivalEvent(ctx, kind, body) + }) + cancel() + if err != nil { + log.Warn().Msg("sync arrival delivery deferred") + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (s *JobsService) deliverArrivalEvent(ctx context.Context, kind models.JobEventType, body any) error { + if _, ok := s.eventHandlers[kind]; !ok { + return fmt.Errorf("durable arrival handler unavailable: %s", kind) + } + return s.HandleEvent(ctx, &models.JobEvent{Type: kind, Body: body}) +} diff --git a/internal/app/consumer/arrival_outbox_live_test.go b/internal/app/consumer/arrival_outbox_live_test.go new file mode 100644 index 000000000..ff4e98c29 --- /dev/null +++ b/internal/app/consumer/arrival_outbox_live_test.go @@ -0,0 +1,151 @@ +package jobs + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/advanced" + "github.com/warmbly/warmbly/internal/app/cipher" + "github.com/warmbly/warmbly/internal/app/warmup" + "github.com/warmbly/warmbly/internal/infrastructure/encryptedkeys" + "github.com/warmbly/warmbly/internal/infrastructure/kms" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +func TestLiveSyncArrivalReplayPreservesHiddenWarmupAndReceiptAuthority(t *testing.T) { + s, d := liveWarmupService(t) + f := newWarmupFixture(t, d) + ctx := t.Context() + f.mintToken(t, s.WarmupRepo) + k, err := kms.NewLocal(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + r := repository.NewDurableEmailMessageMapRepository(d, cipher.NewService(k, nil, encryptedkeys.NewPostgres(d))) + e := f.arrival(f.sentMsgID, nil) + e.UserID = f.senderUser + e.Message.EmailID = f.sender + e.Message.Folder = models.FolderSent + data := repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), MessageID: e.Message.MessageID, ID: e.Message.ID.String(), ThreadID: e.Message.ThreadID} + if err := r.AdmitArrival(ctx, data, &repository.PendingArrival{Arrival: e}); err != nil { + t.Fatal(err) + } + s.InitEvents() + lost := errors.New("consumer terminated after handling") + if err := r.DeliverArrivals(ctx, func(ctx context.Context, kind models.JobEventType, body any) error { + if err := s.deliverArrivalEvent(ctx, kind, body); err != nil { + return err + } + return lost + }); !errors.Is(err, lost) { + t.Fatal(err) + } + if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET retry_at=now() WHERE email_id=$1`, f.sender); err != nil { + t.Fatal(err) + } + fresh, _ := liveWarmupService(t) + fresh.InitEvents() + if err := r.DeliverArrivals(ctx, fresh.deliverArrivalEvent); err != nil { + t.Fatal(err) + } + var visible, pending, receipts int + if err := d.QueryRow(ctx, `SELECT count(*) FROM unibox_emails WHERE id=$1`, e.Message.ID).Scan(&visible); err != nil || visible != 0 { + t.Fatalf("unconfirmed sent copy exposed: %d %v", visible, err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM unibox_pending_emails WHERE id=$1`, e.Message.ID).Scan(&pending); err != nil || pending != 1 { + t.Fatalf("verification deferral lost original ID: %d %v", pending, err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM warmup_received WHERE sender_account_id=$1`, f.sender).Scan(&receipts); err != nil || receipts != 0 { + t.Fatalf("sync invented receipt: %d %v", receipts, err) + } + if err := fresh.HandleEmailSent(ctx, models.SendEmailResult{TaskID: f.task, Success: true, MessageID: f.sentMsgID}); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(ctx, `UPDATE unibox_pending_emails SET retry_at=now() WHERE email_account_id=$1`, f.sender); err != nil { + t.Fatal(err) + } + if err := fresh.retryPendingWarmupVerification(ctx); err != nil { + t.Fatal(err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM unibox_emails WHERE id=$1`, e.Message.ID).Scan(&visible); err != nil || visible != 0 { + t.Fatalf("confirmed warmup exposed: %d %v", visible, err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM warmup_received WHERE sender_account_id=$1`, f.sender).Scan(&receipts); err != nil || receipts != 0 { + t.Fatalf("sent copy invented recipient receipt: %d %v", receipts, err) + } + other := f.arrival("<"+uuid.NewString()+"@test.local>", nil) + other.Message.Subject = "ordinary customer question" + data = repository.EmailMessageData{UserID: other.UserID.String(), EmailID: other.Message.EmailID.String(), MessageID: other.Message.MessageID, ID: other.Message.ID.String()} + if err := r.AdmitArrival(ctx, data, &repository.PendingArrival{Arrival: other}); err != nil { + t.Fatal(err) + } + if err := r.DeliverArrivals(ctx, fresh.deliverArrivalEvent); err != nil { + t.Fatal(err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM unibox_emails WHERE id=$1`, other.Message.ID).Scan(&visible); err != nil || visible != 1 { + t.Fatalf("ordinary arrival missing: %d %v", visible, err) + } +} + +func TestLiveSyncArrivalReportReplayIsIdempotent(t *testing.T) { + d := liveDB(t) + f := newSendResultFixture(t, d) + s := liveJobsService(d) + ctx := t.Context() + task := f.stampSend(t, s) + messageID := "<" + task.String() + "@test.local>" + if _, err := d.Exec(ctx, `UPDATE tasks SET message_id=$2 WHERE id=$1`, task, messageID); err != nil { + t.Fatal(err) + } + s.UniboxRepository = repository.NewUniboxRepository(d) + s.EmailRepository = repository.NewEmailRepostory(d, nil) + s.WarmupRepo = repository.NewWarmupRepository(d.Pool) + s.AdvancedService = advanced.NewService(repository.NewAdvancedOutreachRepository(d.Pool), s.CampaignRepo, s.EmailRepository, s.TaskRepo, repository.NewContactRepostory(d), s.CampaignProgressRepo, nil, nil, nil, nil, warmup.NewService(s.WarmupRepo)) + s.InitEvents() + k, err := kms.NewLocal(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + cipherService := cipher.NewService(k, nil, encryptedkeys.NewPostgres(d)) + r := repository.NewDurableEmailMessageMapRepository(d, cipherService) + id := uuid.New() + recipient := "lead-" + f.contact.String()[:8] + "@test.local" + p := &repository.PendingArrival{Arrival: &models.JobEventNewEmail{UserID: f.user, ReportOriginalMessageID: messageID, Message: &models.EmailMessageStoreData{ID: id, EmailID: f.mailbox, MessageID: "report", Folder: models.FolderInbox}}, Bounce: &models.JobEventInboundBounce{UserID: f.user, EmailID: f.mailbox, OriginalMessageID: messageID, FailedRecipient: recipient, Reason: "recipient refused"}, Complaint: &models.JobEventInboundComplaint{UserID: f.user, EmailID: f.mailbox, OriginalMessageID: messageID, ComplainedRecipient: recipient, Provider: "fixture"}} + data := repository.EmailMessageData{UserID: f.user.String(), EmailID: f.mailbox.String(), MessageID: "report", ID: id.String()} + if err := r.AdmitArrival(ctx, data, p); err != nil { + t.Fatal(err) + } + lost := errors.New("result lost after report was recorded") + for _, target := range []models.JobEventType{models.JobEventTypeInboundBounce, models.JobEventTypeInboundComplaint} { + if err := r.DeliverArrivals(ctx, func(ctx context.Context, kind models.JobEventType, body any) error { + if err := s.deliverArrivalEvent(ctx, kind, body); err != nil { + return err + } + if kind == target { + return lost + } + return nil + }); !errors.Is(err, lost) { + t.Fatal(err) + } + if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET retry_at=now() WHERE email_id=$1`, f.mailbox); err != nil { + t.Fatal(err) + } + } + if err := repository.NewDurableEmailMessageMapRepository(d, cipherService).DeliverArrivals(ctx, s.deliverArrivalEvent); err != nil { + t.Fatal(err) + } + var events, suppressed int + if err := d.QueryRow(ctx, `SELECT count(*) FROM deliverability_events WHERE organization_id=$1`, f.org).Scan(&events); err != nil { + t.Fatal(err) + } + if err := d.QueryRow(ctx, `SELECT count(*) FROM suppressed_recipients WHERE organization_id=$1`, f.org).Scan(&suppressed); err != nil { + t.Fatal(err) + } + if events != 2 || suppressed != 1 { + t.Fatalf("report replay duplicated effects: events=%d suppressed=%d", events, suppressed) + } +} diff --git a/internal/app/consumer/arrival_outbox_test.go b/internal/app/consumer/arrival_outbox_test.go new file mode 100644 index 000000000..d01513822 --- /dev/null +++ b/internal/app/consumer/arrival_outbox_test.go @@ -0,0 +1,21 @@ +package jobs + +import ( + "context" + "testing" + + "github.com/warmbly/warmbly/internal/models" +) + +func TestDurableArrivalRequiresRegisteredHandler(t *testing.T) { + s := &JobsService{} + if err := s.deliverArrivalEvent(context.Background(), models.JobEventTypeNewEmail, &models.JobEventNewEmail{}); err == nil { + t.Fatal("unregistered durable event dropped") + } + called := false + s.InitEvents() + Register(s, models.JobEventTypeNewEmail, func(context.Context, *models.JobEventNewEmail) error { called = true; return nil }) + if err := s.deliverArrivalEvent(context.Background(), models.JobEventTypeNewEmail, &models.JobEventNewEmail{}); err != nil || !called { + t.Fatalf("called=%v err=%v", called, err) + } +} diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index c889872dc..69e7afd7b 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -41,6 +41,7 @@ type JobsService struct { EmailHistoryIDRepository repository.EmailHistoryIDRepository EmailGraphDeltaRepository repository.EmailGraphDeltaRepository EmailSyncStateRepository repository.EmailSyncStateRepository + ArrivalOutbox repository.ArrivalOutbox EmailAccountErrorRepository repository.EmailAccountErrorRepository WarmupRepo repository.WarmupRepository // PoolLinkRepo marks warmup-only mailboxes of linked instances; nil when unused. @@ -126,6 +127,7 @@ func (s *JobsService) Start(ctx context.Context) { // timestamps. go s.sweepFollowUps(ctx) go s.runSendResultEffects(ctx) + go s.runArrivalOutbox(ctx) if err := s.Bus.Subscribe(ctx, []string{kafka.TopicWorkerEvents}, "consumer-group", s.receive); err != nil { log.Error().Err(err).Msg("consumer: worker-events subscription ended") diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index 398f61fc2..7834f62e4 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -745,7 +745,8 @@ var Tables = []Table{ }, { Name: "email_message_map", Group: models.OrgDataGroupInbox, - Scope: `email_id IN ` + orgMailboxes, + Scope: `email_id IN ` + orgMailboxes + ` AND NOT EXISTS (SELECT 1 FROM sync_arrival_outbox o WHERE o.user_id=email_message_map.user_id AND o.email_id=email_message_map.email_id AND o.message_id=email_message_map.message_id AND o.stage=0)`, + Owner: `email_id IN ` + orgMailboxes, Note: "Maps provider message ids to internal ones, so replies still thread after the move.", }, { @@ -994,6 +995,7 @@ var Tables = []Table{ // with the reason. Kept as data so the docs page and the coverage test both // read from one list instead of restating it. var ExcludedTables = map[string]string{ + "sync_arrival_outbox": "Encrypted, org-key-owned arrival delivery queue bound to this instance's provider handles and body storage. Pending maps are excluded too so destination sync retries them; no queue secrets or blob references travel. Full instance backups retain queue, org keys and bodies together.", "send_result_effects": "Source-instance reconciliation delivery queue. Pending webhook and notification effects must drain on the source; imports must not replay customer notifications. Full instance backups retain the durable outbox.", "warmup_received": "Raw receiving-mailbox observations and provider thread handles are source-instance evidence, not portable send authority. Aggregated warmup placement history travels; imported threads cannot continue without new verified observations.", "diagnostic_auth_verifications": "Bounded DKIM observations and grants bind source-instance workers, provider messages and cross-workspace diagnostic parents. They cannot authorize imported mail; destination verification starts unknown. Full instance database backups retain them.", diff --git a/internal/app/orgtransfer/sync_arrival_scope_live_test.go b/internal/app/orgtransfer/sync_arrival_scope_live_test.go new file mode 100644 index 000000000..02823c07e --- /dev/null +++ b/internal/app/orgtransfer/sync_arrival_scope_live_test.go @@ -0,0 +1,63 @@ +package orgtransfer + +import ( + "testing" + + "github.com/google/uuid" +) + +func TestLiveSyncArrivalQueueAndPendingMapsStayInstanceLocal(t *testing.T) { + pool := specLivePool(t) + f := newTenantFixture(t, pool) + ctx := t.Context() + mailbox := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider) VALUES($1,$2,$3,$4,'Arrival','','','smtp_imap')`, mailbox, f.owner, f.org, mailbox.String()+"@test.local"); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM email_accounts WHERE id=$1`, mailbox) }) + for i, key := range []string{"legacy", "pending", "handled"} { + id := uuid.New() + if _, err := pool.Exec(ctx, `INSERT INTO email_message_map(user_id,email_id,message_id,id) VALUES($1,$2,$3,$4)`, f.owner, mailbox, key, id); err != nil { + t.Fatal(err) + } + if i > 0 { + if _, err := pool.Exec(ctx, `INSERT INTO sync_arrival_outbox(user_id,email_id,organization_id,message_id,id,payload,stage) VALUES($1,$2,$3,$4,$5,'fixture',$6)`, f.owner, mailbox, f.org, key, id, i-1); err != nil { + t.Fatal(err) + } + } + } + if ExcludedTables["sync_arrival_outbox"] == "" { + t.Fatal("queue is not explicitly excluded") + } + var scope string + for _, spec := range Tables { + if spec.Name == "sync_arrival_outbox" { + t.Fatal("queue secrets or body references transferred") + } + if spec.Name == "email_message_map" { + scope = spec.Scope + } + } + if scope == "" { + t.Fatal("map scope missing") + } + rows, err := pool.Query(ctx, `SELECT message_id FROM email_message_map WHERE `+scope+` ORDER BY message_id`, f.org) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var keys []string + for rows.Next() { + var key string + if err := rows.Scan(&key); err != nil { + t.Fatal(err) + } + keys = append(keys, key) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if len(keys) != 2 || keys[0] != "handled" || keys[1] != "legacy" { + t.Fatalf("pending maps suppressed destination resync: %v", keys) + } +} diff --git a/internal/app/worker/wmail/admit.go b/internal/app/worker/wmail/admit.go index 745511263..d49c0a65a 100644 --- a/internal/app/worker/wmail/admit.go +++ b/internal/app/worker/wmail/admit.go @@ -2,6 +2,7 @@ package wmail import ( "context" + "errors" "time" "github.com/google/uuid" @@ -164,14 +165,12 @@ func (w *WMail) storeNew(ctx context.Context, msg *models.EmailMessageData, data return err } - if err := w.EmailMessageMapRepository.Add(ctx, repository.EmailMessageData{ + mapping := repository.EmailMessageData{ UserID: w.UserID.String(), EmailID: w.ID.String(), MessageID: mapKey, ID: data.ID.String(), ThreadID: data.ThreadID, - }); err != nil { - return err } // Both report the send they are about, if any. The id goes on the arrival @@ -191,11 +190,22 @@ func (w *WMail) storeNew(ctx context.Context, msg *models.EmailMessageData, data } // The consumer decodes NEW_EMAIL as JobEventNewEmail{user_id, message}. - err := w.onEvent(models.JobEventTypeNewEmail, &models.JobEventNewEmail{ + arrival := &models.JobEventNewEmail{ UserID: w.UserID, Message: data, ReportOriginalMessageID: reportAbout, - }) + } + 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.ErrArrivalOutboxUnsupported) { + return err + } + log.Warn().Str("email_id", w.ID.String()).Msg("sync: legacy arrival publication; upgrade backend and consumer before workers for crash durability") + } + if err := w.EmailMessageMapRepository.Add(ctx, mapping); err != nil { + return err + } + err := w.onEvent(models.JobEventTypeNewEmail, arrival) if err != nil { // The entry would mark a message that never reached the unibox as // known, and every later pass would skip it; drop it so it is re-offered. diff --git a/internal/app/worker/wmail/arrival_durability_test.go b/internal/app/worker/wmail/arrival_durability_test.go new file mode 100644 index 000000000..e2e5bf935 --- /dev/null +++ b/internal/app/worker/wmail/arrival_durability_test.go @@ -0,0 +1,97 @@ +package wmail + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type durableTestMap struct { + recoveryMessageMap + pending map[string]*repository.PendingArrival + failAfterCommit bool + unsupported bool +} + +func (m *durableTestMap) AdmitArrival(ctx context.Context, data repository.EmailMessageData, p *repository.PendingArrival) error { + if m.unsupported { + return repository.ErrArrivalOutboxUnsupported + } + if _, exists := m.data[data.MessageID]; !exists { + m.data[data.MessageID] = data + m.pending[data.MessageID] = p + } + if m.failAfterCommit { + return errors.New("admission response lost") + } + return nil +} + +func TestArrivalDurabilityRestartPreservesIDAcrossProviders(t *testing.T) { + for _, tc := range []struct { + name string + provider models.InboxProvider + store func(*WMail, context.Context, *models.EmailMessageData) error + key string + }{ + {"google", models.InboxProviderGoogle, (*WMail).googleStore, "provider-id"}, + {"graph", models.InboxProviderOutlook, (*WMail).graphStore, "provider-id"}, + {"imap", models.InboxProviderSMTPIMAP, (*WMail).imapStore, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + m := &durableTestMap{recoveryMessageMap: recoveryMessageMap{data: map[string]repository.EmailMessageData{}}, pending: map[string]*repository.PendingArrival{}, failAfterCommit: true} + var events []captured + w := newTestWMail(t, tc.provider, &events) + w.EmailMessageMapRepository = m + msg := &models.EmailMessageData{MessageID: "", GmailID: "provider-id", Subject: "arrival", BodyPlain: "complete body", InReplyTo: []string{""}} + if err := tc.store(w, ctx, msg); err == nil { + t.Fatal("expected ambiguous response") + } + canonical, err := m.Get(ctx, w.UserID, w.ID, tc.key) + if err != nil || canonical == nil { + t.Fatalf("durable mapping missing: %v", err) + } + original := canonical.ID + if len(events) != 0 || len(m.pending) != 1 { + t.Fatal("durable arrival must not depend on worker publication") + } + fresh := newTestWMail(t, tc.provider, &events) + fresh.UserID = w.UserID + fresh.ID = w.ID + fresh.EmailMessageMapRepository = m + m.failAfterCommit = false + if !fresh.retryUnmap(ctx) { + t.Fatal("fresh worker failed") + } + // Re-offering after a lost response cannot replace the queued canonical ID. + if err := tc.store(fresh, ctx, msg); err != nil { + t.Fatal(err) + } + p := m.pending[tc.key] + if len(m.pending) != 1 || p.Arrival.Message.ID.String() != original || m.data[tc.key].ID != original || len(events) != 0 { + t.Fatal("restart reminted the arrival or published outside durable delivery") + } + if p.Arrival.UserID != w.UserID || p.Arrival.Message.EmailID != w.ID || p.Arrival.Message.ID == uuid.Nil { + t.Fatal("ownership not retained") + } + }) + } +} + +func TestArrivalOldBackendKeepsLegacyPublication(t *testing.T) { + var events []captured + w := newTestWMail(t, models.InboxProviderGoogle, &events) + m := &durableTestMap{recoveryMessageMap: recoveryMessageMap{data: map[string]repository.EmailMessageData{}}, unsupported: true} + w.EmailMessageMapRepository = m + if err := w.googleStore(context.Background(), &models.EmailMessageData{MessageID: "legacy", GmailID: "legacy"}); err != nil { + t.Fatal(err) + } + if len(newEmails(events)) != 1 || len(m.data) != 1 { + t.Fatal("old backend lost existing sync functionality") + } +} diff --git a/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.down.sql b/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.down.sql new file mode 100644 index 000000000..f3f03ceac --- /dev/null +++ b/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.down.sql @@ -0,0 +1,8 @@ +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM sync_arrival_outbox) THEN + RAISE EXCEPTION 'Cannot downgrade with unresolved sync arrival delivery'; + END IF; +END $$; + +DROP TABLE sync_arrival_outbox; diff --git a/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.up.sql b/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.up.sql new file mode 100644 index 000000000..ff3ddb279 --- /dev/null +++ b/internal/infrastructure/db/migrations/000278_sync_arrival_outbox.up.sql @@ -0,0 +1,18 @@ +CREATE TABLE sync_arrival_outbox ( + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + email_id uuid NOT NULL REFERENCES email_accounts(id) ON DELETE CASCADE, + organization_id uuid NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + message_id text NOT NULL, + id uuid NOT NULL, + payload text NOT NULL, + stage smallint NOT NULL DEFAULT 0 CHECK (stage BETWEEN 0 AND 3), + lease uuid, + locked_until timestamptz, + retry_at timestamptz NOT NULL DEFAULT now(), + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (user_id, email_id, message_id), + FOREIGN KEY (user_id, email_id, message_id) + REFERENCES email_message_map(user_id, email_id, message_id) ON DELETE CASCADE +); +CREATE INDEX sync_arrival_outbox_due ON sync_arrival_outbox(retry_at, created_at); +CREATE INDEX sync_arrival_outbox_pending ON sync_arrival_outbox(user_id, email_id, id) WHERE stage = 0; diff --git a/internal/infrastructure/db/sync_arrival_upgrade_test.go b/internal/infrastructure/db/sync_arrival_upgrade_test.go new file mode 100644 index 000000000..62f287c31 --- /dev/null +++ b/internal/infrastructure/db/sync_arrival_upgrade_test.go @@ -0,0 +1,106 @@ +package db + +import ( + "context" + "net/url" + "os" + "strings" + "testing" + + "github.com/golang-migrate/migrate/v4/source/iofs" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +func TestLiveSyncArrivalUpgradePreservesLegacyMaps(t *testing.T) { + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + u, err := url.Parse(dsn) + if err != nil || (u.Hostname() != "127.0.0.1" && u.Hostname() != "localhost") { + t.Fatal("upgrade fixture requires local PostgreSQL") + } + ctx := t.Context() + admin, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer admin.Close(context.Background()) + name := "arrival_upgrade_" + uuid.NewString()[:8] + quoted := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE DATABASE "+quoted); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil { + t.Error(err) + } + }() + u.Path = "/" + name + source, err := iofs.New(migrationsFS, "migrations") + if err != nil { + t.Fatal(err) + } + m, err := openMigrate(source, u.String()) + if err != nil { + t.Fatal(err) + } + defer m.Close() + if err := m.Migrate(277); err != nil { + t.Fatal(err) + } + conn, err := pgx.Connect(ctx, u.String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + user, org, mailbox, id := uuid.New(), uuid.New(), uuid.New(), uuid.New() + for _, row := range []struct { + sql string + args []any + }{ + {`INSERT INTO users(id,email,first_name,last_name)VALUES($1,$2,'Legacy','Fixture')`, []any{user, user.String() + "@test.local"}}, + {`INSERT INTO organizations(id,name,slug,owner_user_id)VALUES($1,'Legacy',$2,$3)`, []any{org, org.String(), user}}, + {`INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider)VALUES($1,$2,$3,$4,'Legacy','','','smtp_imap')`, []any{mailbox, user, org, mailbox.String() + "@test.local"}}, + {`INSERT INTO email_sync_state(email_id,user_id,backfill_status,backfill_synced)VALUES($1,$2,'complete',42)`, []any{mailbox, user}}, + {`INSERT INTO email_message_map(user_id,email_id,message_id,id,thread_id)VALUES($1,$2,'legacy',$3,'thread')`, []any{user, mailbox, id}}, + } { + if _, err := conn.Exec(ctx, row.sql, row.args...); err != nil { + t.Fatal(err) + } + } + if err := m.Up(); err != nil { + t.Fatal(err) + } + var retained bool + if err := conn.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM email_message_map m JOIN email_sync_state s ON s.email_id=m.email_id WHERE m.id=$1 AND s.backfill_status='complete' AND s.backfill_synced=42 AND m.thread_id='thread')`, id).Scan(&retained); err != nil || !retained { + t.Fatal("upgrade changed legacy map or completed import", err) + } + var count int + if err := conn.QueryRow(ctx, `SELECT count(*) FROM sync_arrival_outbox`).Scan(&count); err != nil || count != 0 { + t.Fatal("upgrade invented pending arrivals", err) + } + if _, err := conn.Exec(ctx, `INSERT INTO sync_arrival_outbox(user_id,email_id,organization_id,message_id,id,payload)VALUES($1,$2,$3,'legacy',$4,'fixture')`, user, mailbox, org, id); err != nil { + t.Fatal(err) + } + down, err := migrationsFS.ReadFile("migrations/000278_sync_arrival_outbox.down.sql") + if err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(ctx, string(down)); err == nil || !strings.Contains(err.Error(), "unresolved sync arrival") { + t.Fatalf("downgrade discarded pending delivery: %v", err) + } + if _, err := conn.Exec(ctx, `DELETE FROM sync_arrival_outbox`); err != nil { + t.Fatal(err) + } + if err := m.Steps(-1); err != nil { + t.Fatal(err) + } + if err := conn.QueryRow(ctx, `SELECT count(*) FROM email_message_map WHERE id=$1`, id).Scan(&count); err != nil || count != 1 { + t.Fatal("downgrade removed canonical map", err) + } + if err := m.Up(); err != nil { + t.Fatal(err) + } +} diff --git a/internal/infrastructure/kafka/consumer.go b/internal/infrastructure/kafka/consumer.go index 90054151d..2a1d9a10e 100644 --- a/internal/infrastructure/kafka/consumer.go +++ b/internal/infrastructure/kafka/consumer.go @@ -128,14 +128,31 @@ func (cons *Consumer) Consume(ctx context.Context, handler func(msg *ckf.Message } // Log transient Kafka errors and retry after a brief delay log.Warn().Str("code", fmt.Sprintf("%d", kafkaErr.Code())).Err(kafkaErr).Msg("kafka consumer error") - time.Sleep(time.Second) + if err := consumerBackoff(ctx, time.Second); err != nil { + return err + } continue } return fmt.Errorf("error reading message: %w", err) } - if err := handler(msg); err != nil { - log.Error().Err(err).Msg("kafka message handler error") + backoff := 100 * time.Millisecond + for { + if err := handler(msg); err == nil { + break + } else { + log.Error().Err(err).Msg("kafka message handler error; retrying before advancing offsets") + } + if err := consumerBackoff(ctx, backoff); err != nil { + return err + } + cons.mu.Lock() + closed := cons.closed + cons.mu.Unlock() + if closed { + return ErrClientClosed + } + backoff = min(2*backoff, 5*time.Second) } // Needs enable.auto.offset.store=false; the background commit sends it. @@ -143,8 +160,19 @@ func (cons *Consumer) Consume(ctx context.Context, handler func(msg *ckf.Message if errors.Is(err, ErrClientClosed) { return err } - log.Warn().Err(err).Msg("kafka: storing a handled offset failed") + return fmt.Errorf("kafka: storing a handled offset failed: %w", err) } } } } + +func consumerBackoff(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/internal/infrastructure/kafka/consumer_retry_test.go b/internal/infrastructure/kafka/consumer_retry_test.go new file mode 100644 index 000000000..88d48667e --- /dev/null +++ b/internal/infrastructure/kafka/consumer_retry_test.go @@ -0,0 +1,85 @@ +//go:build kafka + +package kafka + +import ( + "context" + "errors" + "testing" + "time" + + ckf "github.com/confluentinc/confluent-kafka-go/v2/kafka" +) + +type orderedConsumer struct { + shutdownConsumer + queue []*ckf.Message + stored []ckf.Offset + reads int + storeErr error +} + +func (c *orderedConsumer) ReadMessage(time.Duration) (*ckf.Message, error) { + c.reads++ + if len(c.queue) == 0 { + return nil, errors.New("unexpected poll") + } + m := c.queue[0] + c.queue = c.queue[1:] + return m, nil +} +func (c *orderedConsumer) StoreMessage(m *ckf.Message) ([]ckf.TopicPartition, error) { + c.stored = append(c.stored, m.TopicPartition.Offset) + return nil, c.storeErr +} + +func TestConsumerRetriesFailedRecordBeforeLaterOffsets(t *testing.T) { + first := &ckf.Message{TopicPartition: ckf.TopicPartition{Partition: 0, Offset: 10}} + second := &ckf.Message{TopicPartition: ckf.TopicPartition{Partition: 0, Offset: 11}} + c := &orderedConsumer{queue: []*ckf.Message{first, second}} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var handled []ckf.Offset + err := (&Consumer{c: c}).Consume(ctx, func(m *ckf.Message) error { + handled = append(handled, m.TopicPartition.Offset) + if len(handled) == 1 { + return errors.New("transient database failure") + } + if len(handled) == 2 && (c.reads != 1 || len(c.stored) != 0 || m != first) { + t.Fatal("failed record was skipped or acked") + } + if m == second { + cancel() + } + return nil + }) + if !errors.Is(err, context.Canceled) || len(handled) != 3 || handled[0] != 10 || handled[1] != 10 || handled[2] != 11 || len(c.stored) != 2 || c.stored[0] != 10 || c.stored[1] != 11 { + t.Fatalf("err=%v handled=%v stored=%v", err, handled, c.stored) + } +} + +func TestConsumerFailureCancellationNeverStoresOffset(t *testing.T) { + c := &orderedConsumer{queue: []*ckf.Message{{}}} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + err := (&Consumer{c: c}).Consume(ctx, func(*ckf.Message) error { cancel(); return errors.New("failed") }) + if !errors.Is(err, context.Canceled) || len(c.stored) != 0 || c.reads != 1 { + t.Fatalf("err=%v stored=%v reads=%d", err, c.stored, c.reads) + } +} + +func TestConsumerOffsetStorageFailureStopsBeforeNextRecord(t *testing.T) { + c := &orderedConsumer{queue: []*ckf.Message{{}, {}}, storeErr: errors.New("store unavailable")} + err := (&Consumer{c: c}).Consume(context.Background(), func(*ckf.Message) error { return nil }) + if !errors.Is(err, c.storeErr) || c.reads != 1 { + t.Fatalf("err=%v reads=%d", err, c.reads) + } +} + +func TestConsumerRetryBackoffCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := consumerBackoff(ctx, 5*time.Second); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } +} diff --git a/internal/repository/email_message_map.go b/internal/repository/email_message_map.go index 863d2128b..8f1f9f79d 100644 --- a/internal/repository/email_message_map.go +++ b/internal/repository/email_message_map.go @@ -2,8 +2,10 @@ package repository import ( "context" + "errors" "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" ) // EmailMessageMapRepository maps a provider messageId to the internal email @@ -20,6 +22,24 @@ type EmailMessageMapRepository interface { Del(ctx context.Context, userID, emailID uuid.UUID, messageID string, id uuid.UUID) error } +var ErrArrivalOutboxUnsupported = errors.New("durable arrival admission unavailable; upgrade backend and consumer first") + +// ArrivalAdmission is an optional internal protocol, separate from legacy map writes. +type ArrivalAdmission interface { + AdmitArrival(context.Context, EmailMessageData, *PendingArrival) error +} + +type PendingArrival struct { + Arrival *models.JobEventNewEmail `json:"arrival"` + Bounce *models.JobEventInboundBounce `json:"bounce,omitempty"` + Complaint *models.JobEventInboundComplaint `json:"complaint,omitempty"` +} + +type ArrivalOutbox interface { + HasPendingArrival(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) (bool, error) + DeliverArrivals(context.Context, func(context.Context, models.JobEventType, any) error) error +} + // EmailMessageData is one (user, mailbox, providerMessageID) -> internal // (id, threadID) mapping row. The UUID-typed values travel as strings to match // the worker call sites, which pass uuid.String(). diff --git a/internal/repository/http_email_message_map.go b/internal/repository/http_email_message_map.go index e9739fde2..0c9e6f2d1 100644 --- a/internal/repository/http_email_message_map.go +++ b/internal/repository/http_email_message_map.go @@ -93,6 +93,34 @@ func (r *httpEmailMessageMapRepository) Add(ctx context.Context, data EmailMessa } } +func (r *httpEmailMessageMapRepository) AdmitArrival(ctx context.Context, data EmailMessageData, pending *PendingArrival) error { + body, err := json.Marshal(struct { + Map emailMessageMapPayload `json:"map"` + Pending *PendingArrival `json:"pending"` + }{emailMessageMapPayload(data), pending}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.endpoint()+"/arrival", bytes.NewReader(body)) + if err != nil { + return err + } + r.authed(req) + req.Header.Set("Content-Type", "application/json") + resp, err := r.client.Do(req) + if err != nil { + return fmt.Errorf("arrival admission: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusMethodNotAllowed { + return ErrArrivalOutboxUnsupported + } + if resp.StatusCode != http.StatusNoContent || resp.Header.Get("X-Warmbly-Arrival-Durable") != "1" { + return fmt.Errorf("arrival admission not confirmed: status %d", resp.StatusCode) + } + return nil +} + func (r *httpEmailMessageMapRepository) Get(ctx context.Context, userID, emailID uuid.UUID, messageID string) (*EmailMessageData, error) { q := url.Values{} q.Set("user_id", userID.String()) diff --git a/internal/repository/http_sync_arrival_test.go b/internal/repository/http_sync_arrival_test.go new file mode 100644 index 000000000..4853fcae5 --- /dev/null +++ b/internal/repository/http_sync_arrival_test.go @@ -0,0 +1,67 @@ +package repository + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +func TestHTTPArrivalAdmissionUpgradeAcknowledgement(t *testing.T) { + for _, tc := range []struct { + name string + status int + ack string + unsupported, success bool + }{ + {"old-backend", 404, "", true, false}, {"old-method", 405, "", true, false}, + {"durable", 204, "1", false, true}, {"unconfirmed", 204, "", false, false}, + {"backend-failure", 503, "", false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/internal/email-message-map/arrival" || r.Method != http.MethodPost || r.Header.Get("Authorization") != "Bearer fixture-token" { + t.Error("incorrect internal protocol") + } + 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(context.Background(), EmailMessageData{}, &PendingArrival{}) + if errors.Is(err, ErrArrivalOutboxUnsupported) != tc.unsupported || (err == nil) != tc.success { + t.Fatalf("err=%v", err) + } + }) + } +} + +func TestHTTPArrivalAdmissionResponseLossDoesNotDowngrade(t *testing.T) { + var committed atomic.Bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Error("ambiguous admission fell back to legacy write") + } + committed.Store(true) + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + _ = conn.Close() + })) + defer srv.Close() + r, err := NewHTTPEmailMessageMapRepository(srv.URL, "fixture-token") + if err != nil { + t.Fatal(err) + } + err = r.(ArrivalAdmission).AdmitArrival(context.Background(), EmailMessageData{}, &PendingArrival{}) + if err == nil || errors.Is(err, ErrArrivalOutboxUnsupported) || !committed.Load() { + t.Fatalf("response loss treated as unsupported: committed=%v err=%v", committed.Load(), err) + } +} diff --git a/internal/repository/pg_email_message_map.go b/internal/repository/pg_email_message_map.go index 0225a6cb5..74e726657 100644 --- a/internal/repository/pg_email_message_map.go +++ b/internal/repository/pg_email_message_map.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/warmbly/warmbly/internal/app/cipher" "github.com/warmbly/warmbly/internal/infrastructure/db" ) @@ -14,7 +15,8 @@ import ( // backend, which serves it to workers over the internal API. The composite key // (user_id, email_id, message_id) is canonical — see the 000003 migration. type pgEmailMessageMapRepository struct { - db *db.DB + db *db.DB + cipher cipher.CipherService } // NewEmailMessageMapRepository returns the Postgres-backed implementation. @@ -42,6 +44,8 @@ func (r *pgEmailMessageMapRepository) Add(ctx context.Context, data EmailMessage VALUES ($1, $2, $3, $4, $5) ON CONFLICT (user_id, email_id, message_id) DO UPDATE SET id = EXCLUDED.id, thread_id = EXCLUDED.thread_id + WHERE NOT EXISTS (SELECT 1 FROM sync_arrival_outbox o + WHERE o.user_id = EXCLUDED.user_id AND o.email_id = EXCLUDED.email_id AND o.message_id = EXCLUDED.message_id) ` if _, err := r.db.Exec(ctx, q, userID, emailID, data.MessageID, id, data.ThreadID); err != nil { return fmt.Errorf("email_message_map: add: %w", err) @@ -73,11 +77,13 @@ func (r *pgEmailMessageMapRepository) Get(ctx context.Context, userID, emailID u }, nil } -func (r *pgEmailMessageMapRepository) Del(ctx context.Context, userID, emailID uuid.UUID, messageID string, _ uuid.UUID) error { - // (user_id, email_id, message_id) is the primary key, so the internal id - // is not needed to identify the row. - const q = `DELETE FROM email_message_map WHERE user_id = $1 AND email_id = $2 AND message_id = $3` - if _, err := r.db.Exec(ctx, q, userID, emailID, messageID); err != nil { +func (r *pgEmailMessageMapRepository) Del(ctx context.Context, userID, emailID uuid.UUID, messageID string, id uuid.UUID) error { + // Legacy publication rollback must not remove an already admitted arrival. + const q = `DELETE FROM email_message_map m WHERE user_id=$1 AND email_id=$2 AND message_id=$3 + AND ($4::uuid='00000000-0000-0000-0000-000000000000' OR id=$4) + AND NOT EXISTS (SELECT 1 FROM sync_arrival_outbox o + WHERE o.user_id=m.user_id AND o.email_id=m.email_id AND o.message_id=m.message_id)` + if _, err := r.db.Exec(ctx, q, userID, emailID, messageID, id); err != nil { return fmt.Errorf("email_message_map: del: %w", err) } return nil diff --git a/internal/repository/pg_sync_arrival.go b/internal/repository/pg_sync_arrival.go new file mode 100644 index 000000000..727cf8677 --- /dev/null +++ b/internal/repository/pg_sync_arrival.go @@ -0,0 +1,176 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/warmbly/warmbly/internal/app/cipher" + "github.com/warmbly/warmbly/internal/infrastructure/db" + "github.com/warmbly/warmbly/internal/models" +) + +func NewDurableEmailMessageMapRepository(d *db.DB, c cipher.CipherService) *pgEmailMessageMapRepository { + return &pgEmailMessageMapRepository{db: d, cipher: c} +} + +func validateArrival(data EmailMessageData, pending *PendingArrival) (uuid.UUID, uuid.UUID, uuid.UUID, error) { + user, uerr := uuid.Parse(data.UserID) + email, eerr := uuid.Parse(data.EmailID) + id, ierr := uuid.Parse(data.ID) + if uerr != nil || eerr != nil || ierr != nil || user == uuid.Nil || email == uuid.Nil || id == uuid.Nil || data.MessageID == "" { + return user, email, id, errors.New("invalid arrival mapping") + } + if pending == nil || pending.Arrival == nil || pending.Arrival.Message == nil || + pending.Arrival.UserID != user || pending.Arrival.Message.EmailID != email || + pending.Arrival.Message.ID != id || pending.Arrival.Message.ThreadID != data.ThreadID { + return user, email, id, errors.New("arrival does not match mapping") + } + if pending.Bounce != nil && (pending.Bounce.UserID != user || pending.Bounce.EmailID != email) || + pending.Complaint != nil && (pending.Complaint.UserID != user || pending.Complaint.EmailID != email) { + return user, email, id, errors.New("arrival report does not match mailbox") + } + return user, email, id, nil +} + +func (r *pgEmailMessageMapRepository) AdmitArrival(ctx context.Context, data EmailMessageData, pending *PendingArrival) error { + user, email, id, err := validateArrival(data, pending) + if err != nil { + return err + } + if r.cipher == nil { + return ErrArrivalOutboxUnsupported + } + 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 fmt.Errorf("arrival mailbox ownership: %w", err) + } + sealed, err := r.cipher.Cipher(ctx, org) + if err != nil { + return err + } + raw, err := json.Marshal(pending) + if err != nil { + return err + } + payload, err := sealed.Encrypt(ctx, string(raw)) + if err != nil { + return err + } + tx, err := r.db.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(context.Background()) + // 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 fmt.Errorf("arrival mailbox ownership: %w", 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) + VALUES($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING`, user, email, data.MessageID, id, data.ThreadID) + if err != nil { + return err + } + if inserted.RowsAffected() != 0 { + _, err = tx.Exec(ctx, `INSERT INTO sync_arrival_outbox(user_id,email_id,organization_id,message_id,id,payload) + VALUES($1,$2,$3,$4,$5,$6)`, user, email, org, data.MessageID, id, payload) + if err != nil { + return err + } + } + return tx.Commit(ctx) +} + +func (r *pgEmailMessageMapRepository) HasPendingArrival(ctx context.Context, user, email, id uuid.UUID) (bool, error) { + var pending bool + err := r.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM sync_arrival_outbox o + JOIN email_accounts a ON a.id=o.email_id AND a.user_id=o.user_id AND a.organization_id=o.organization_id + WHERE o.user_id=$1 AND o.email_id=$2 AND o.id=$3 AND o.stage=0)`, user, email, id).Scan(&pending) + return pending, err +} + +// DeliverArrivals invokes the existing consumer handlers, retaining every failed stage. +func (r *pgEmailMessageMapRepository) DeliverArrivals(ctx context.Context, deliver func(context.Context, models.JobEventType, any) error) error { + var first error + for i := 0; i < 32; i++ { + found, err := r.deliverArrival(ctx, deliver) + if err != nil && first == nil { + first = err + } + if !found || ctx.Err() != nil { + break + } + } + return first +} + +func (r *pgEmailMessageMapRepository) deliverArrival(ctx context.Context, deliver func(context.Context, models.JobEventType, any) error) (bool, error) { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + var user, email, org, lease uuid.UUID + var key, payload string + var stage int + // Claim without retaining a connection while handlers use their repositories. + err := r.db.QueryRow(ctx, `WITH candidate AS (SELECT o.user_id,o.email_id,o.message_id + FROM sync_arrival_outbox o JOIN email_accounts a ON a.id=o.email_id AND a.user_id=o.user_id AND a.organization_id=o.organization_id + WHERE o.retry_at<=now() AND (o.locked_until IS NULL OR o.locked_until"} + p.Complaint = &models.JobEventInboundComplaint{UserID: f.user, EmailID: f.mailbox, OriginalMessageID: ""} + wrong := uuid.New() + data.UserID = wrong.String() + p.Arrival.UserID = wrong + p.Bounce.UserID = wrong + p.Complaint.UserID = wrong + if err := r.AdmitArrival(ctx, data, p); err == nil { + t.Fatal("another user admitted mail") + } + data.UserID = f.user.String() + p.Arrival.UserID = f.user + p.Bounce.UserID = f.user + p.Complaint.UserID = f.user + p.Bounce.EmailID = uuid.New() + if err := r.AdmitArrival(ctx, data, p); err == nil { + t.Fatal("cross-mailbox report accepted") + } + p.Bounce.EmailID = f.mailbox + if err := r.AdmitArrival(ctx, data, p); err != nil { + t.Fatal(err) + } + var seen []models.JobEventType + failed := errors.New("report handler unavailable") + handle := func(_ context.Context, kind models.JobEventType, _ any) error { + seen = append(seen, kind) + if kind == models.JobEventTypeInboundBounce && len(seen) == 2 { + return failed + } + return nil + } + if err := r.DeliverArrivals(ctx, handle); !errors.Is(err, failed) { + t.Fatal(err) + } + if pending, err := r.HasPendingArrival(ctx, f.user, f.mailbox, p.Arrival.Message.ID); err != nil || pending { + t.Fatal("completed arrival confused with pending report") + } + dueArrival(t, d, f) + if err := NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d)).DeliverArrivals(ctx, handle); err != nil { + t.Fatal(err) + } + if len(seen) != 4 || seen[0] != models.JobEventTypeNewEmail || seen[1] != models.JobEventTypeInboundBounce || seen[2] != models.JobEventTypeInboundBounce || seen[3] != models.JobEventTypeInboundComplaint { + t.Fatalf("stages=%v", seen) + } + // Completed and legacy mappings are not guessed to be missing arrivals. + legacy, lp := arrivalLivePayload(f, "legacy") + if err := r.Add(ctx, legacy); err != nil { + t.Fatal(err) + } + lp.Arrival.Message.ID = uuid.New() + retry := legacy + retry.ID = lp.Arrival.Message.ID.String() + if err := r.AdmitArrival(ctx, retry, lp); err != nil { + t.Fatal(err) + } + known, err := r.Get(ctx, f.user, f.mailbox, legacy.MessageID) + if err != nil || known.ID != legacy.ID { + t.Fatal("legacy mapping replaced") + } + if err := r.DeliverArrivals(ctx, func(context.Context, models.JobEventType, any) error { + t.Fatal("legacy or completed mapping replayed") + return nil + }); err != nil { + t.Fatal(err) + } +} + +func TestLiveSyncArrivalAtomicRollbackDeletedMailboxAndLease(t *testing.T) { + d := liveUniboxFolderDB(t) + f := newUniboxFolderFixture(t, d.Pool) + ctx := context.Background() + r := NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d)) + data, p := arrivalLivePayload(f, "atomic") + // Force only the outbox insert to fail, without disturbing ordinary mappings. + if _, err := d.Exec(ctx, `ALTER TABLE sync_arrival_outbox ADD CONSTRAINT sync_arrival_test_reject CHECK (message_id<>'atomic')`); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, _ = d.Exec(ctx, `ALTER TABLE sync_arrival_outbox DROP CONSTRAINT IF EXISTS sync_arrival_test_reject`) + }) + if err := r.AdmitArrival(ctx, data, p); err == nil { + t.Fatal("expected outbox insertion failure") + } + if known, err := r.Get(ctx, f.user, f.mailbox, data.MessageID); err != nil || known != nil { + t.Fatal("map escaped rolled-back admission") + } + if _, err := d.Exec(ctx, `ALTER TABLE sync_arrival_outbox DROP CONSTRAINT sync_arrival_test_reject`); err != nil { + t.Fatal(err) + } + if err := r.AdmitArrival(ctx, data, p); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET locked_until=now()+interval '1 minute' WHERE email_id=$1`, f.mailbox); err != nil { + t.Fatal(err) + } + if err := r.DeliverArrivals(ctx, func(context.Context, models.JobEventType, any) error { t.Fatal("active lease replayed"); return nil }); err != nil { + t.Fatal(err) + } + dueArrival(t, d, f) + // A process lost while leased becomes replayable, including on a one-connection self-host. + config := d.Pool.Config() + config.MaxConns = 1 + config.MinConns = 0 + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + single := &db.DB{Pool: pool} + sr := NewDurableEmailMessageMapRepository(single, arrivalLiveCipher(t, single)) + if err := sr.DeliverArrivals(ctx, func(ctx context.Context, _ models.JobEventType, _ any) error { + var n int + return single.QueryRow(ctx, "SELECT 1").Scan(&n) + }); err != nil { + t.Fatal(err) + } + data, p = arrivalLivePayload(f, "single-connection") + if err := sr.AdmitArrival(ctx, data, p); err != nil { + t.Fatal(err) + } + if err := sr.DeliverArrivals(ctx, func(context.Context, models.JobEventType, any) error { return nil }); err != nil { + t.Fatal(err) + } + data, p = arrivalLivePayload(f, "deleted") + if err := r.AdmitArrival(ctx, data, p); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(ctx, `DELETE FROM email_accounts WHERE id=$1`, f.mailbox); err != nil { + t.Fatal(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") + } + var n int + if err := d.QueryRow(ctx, `SELECT count(*) FROM sync_arrival_outbox WHERE email_id=$1`, f.mailbox).Scan(&n); err != nil || n != 0 { + t.Fatal("deleted mailbox retained queue") + } +}