feat: atomically retain encrypted mailbox arrivals for original-ID consumer replay and retry Kafka handlers before advancing offsets

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.

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),

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)
+	}
+}

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).

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

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})
+}

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)
+	}
+}

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)
+	}
+}

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")

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.",

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)
+	}
+}

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.

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, "<arrival@test.local>"},
+	} {
+		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: "<arrival@test.local>", GmailID: "provider-id", Subject: "arrival", BodyPlain: "complete body", InReplyTo: []string{"<send@test.local>"}}
+			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")
+	}
+}

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;

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;

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)
+	}
+}

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
+	}
+}

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)
+	}
+}

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().

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())

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)
+	}
+}

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

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<now())
+		ORDER BY o.retry_at,o.created_at LIMIT 1 FOR UPDATE OF o SKIP LOCKED)
+		UPDATE sync_arrival_outbox o SET lease=gen_random_uuid(),locked_until=now()+interval '1 minute'
+		FROM candidate c WHERE o.user_id=c.user_id AND o.email_id=c.email_id AND o.message_id=c.message_id
+		RETURNING o.user_id,o.email_id,o.organization_id,o.message_id,o.payload,o.stage,o.lease`).Scan(&user, &email, &org, &key, &payload, &stage, &lease)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	sealed, err := r.cipher.Cipher(ctx, org)
+	var raw string
+	if err == nil {
+		raw, err = sealed.Decrypt(ctx, payload)
+	}
+	var pending PendingArrival
+	if err == nil {
+		err = json.Unmarshal([]byte(raw), &pending)
+	}
+	if err == nil {
+		events := []struct {
+			kind    models.JobEventType
+			body    any
+			present bool
+		}{
+			{models.JobEventTypeNewEmail, pending.Arrival, true},
+			{models.JobEventTypeInboundBounce, pending.Bounce, pending.Bounce != nil},
+			{models.JobEventTypeInboundComplaint, pending.Complaint, pending.Complaint != nil},
+		}
+		for stage < len(events) {
+			e := events[stage]
+			if e.present {
+				if err = deliver(ctx, e.kind, e.body); err != nil {
+					break
+				}
+			}
+			stage++
+			if _, updateErr := r.db.Exec(ctx, `UPDATE sync_arrival_outbox SET stage=$5
+				WHERE user_id=$1 AND email_id=$2 AND message_id=$3 AND lease=$4`, user, email, key, lease, stage); updateErr != nil {
+				return true, updateErr
+			}
+		}
+	}
+	if err == nil {
+		_, err = r.db.Exec(ctx, `DELETE FROM sync_arrival_outbox WHERE user_id=$1 AND email_id=$2 AND message_id=$3 AND lease=$4`, user, email, key, lease)
+	} else {
+		if _, updateErr := r.db.Exec(ctx, `UPDATE sync_arrival_outbox SET locked_until=NULL,retry_at=now()+interval '15 seconds'
+			WHERE user_id=$1 AND email_id=$2 AND message_id=$3 AND lease=$4`, user, email, key, lease); updateErr != nil {
+			return true, updateErr
+		}
+	}
+	return true, err
+}

internal/repository/pg_sync_arrival_live_test.go

@@ -0,0 +1,276 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/app/cipher"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/infrastructure/encryptedkeys"
+	"github.com/warmbly/warmbly/internal/infrastructure/kms"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func arrivalLiveCipher(t *testing.T, d *db.DB) cipher.CipherService {
+	t.Helper()
+	k, err := kms.NewLocal(make([]byte, 32))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return cipher.NewService(k, nil, encryptedkeys.NewPostgres(d))
+}
+
+func arrivalLivePayload(f *uniboxFolderFixture, key string) (EmailMessageData, *PendingArrival) {
+	id := uuid.New()
+	now := time.Now().UTC()
+	return EmailMessageData{UserID: f.user.String(), EmailID: f.mailbox.String(), MessageID: key, ID: id.String(), ThreadID: key}, &PendingArrival{Arrival: &models.JobEventNewEmail{UserID: f.user, Message: &models.EmailMessageStoreData{ID: id, EmailID: f.mailbox, MessageID: key, ThreadID: key, Folder: models.FolderInbox, Subject: "encrypted durable arrival", BodyText: "protected content", InternalDate: now, SentDate: now, CreatedAt: now, UpdatedAt: now}}}
+}
+
+func dueArrival(t *testing.T, d *db.DB, f *uniboxFolderFixture) {
+	t.Helper()
+	if _, err := d.Exec(context.Background(), `UPDATE sync_arrival_outbox SET retry_at=now(),locked_until=NULL WHERE email_id=$1`, f.mailbox); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestLiveSyncArrivalRestartAndAmbiguousAdmission(t *testing.T) {
+	d := liveUniboxFolderDB(t)
+	f := newUniboxFolderFixture(t, d.Pool)
+	ctx := context.Background()
+	c := arrivalLiveCipher(t, d)
+	r := NewDurableEmailMessageMapRepository(d, c)
+	if _, err := d.Exec(ctx, `INSERT INTO email_sync_state(email_id,user_id,backfill_status,backfill_synced) VALUES($1,$2,'complete',42)`, f.mailbox, f.user); err != nil {
+		t.Fatal(err)
+	}
+	data, p := arrivalLivePayload(f, "restart")
+	if err := r.AdmitArrival(ctx, data, p); err != nil {
+		t.Fatal(err)
+	}
+	var payload string
+	if err := d.QueryRow(ctx, `SELECT payload FROM sync_arrival_outbox WHERE email_id=$1`, f.mailbox).Scan(&payload); err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(payload, p.Arrival.Message.Subject) || strings.Contains(payload, "protected content") {
+		t.Fatal("outbox plaintext persisted")
+	}
+	other := newUniboxFolderFixture(t, d.Pool)
+	otherCipher, err := c.Cipher(ctx, other.org)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := otherCipher.Decrypt(ctx, payload); err == nil {
+		t.Fatal("another organization decrypted pending mail")
+	}
+	// Fresh repository and key cache after admission, before delivery.
+	r = NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d))
+	original := p.Arrival.Message.ID
+	p.Arrival.Message.ID = uuid.New()
+	data.ID = p.Arrival.Message.ID.String()
+	if err := r.AdmitArrival(ctx, data, p); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Add(ctx, data); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Del(ctx, f.user, f.mailbox, data.MessageID, p.Arrival.Message.ID); err != nil {
+		t.Fatal(err)
+	}
+	known, err := r.Get(ctx, f.user, f.mailbox, data.MessageID)
+	if err != nil || known == nil || known.ID != original.String() {
+		t.Fatalf("canonical ID changed: %v %v", known, err)
+	}
+	if pending, err := r.HasPendingArrival(ctx, f.user, f.mailbox, original); err != nil || !pending {
+		t.Fatalf("pending=%v err=%v", pending, err)
+	}
+	if pending, err := r.HasPendingArrival(ctx, other.user, f.mailbox, original); err != nil || pending {
+		t.Fatal("pending marker leaked ownership")
+	}
+	unibox := NewUniboxRepository(d)
+	calls := 0
+	lost := errors.New("consumer result lost after insert")
+	handle := func(ctx context.Context, kind models.JobEventType, body any) error {
+		if kind != models.JobEventTypeNewEmail {
+			t.Fatal(kind)
+		}
+		e := body.(*models.JobEventNewEmail)
+		if e.Message.ID != original || e.UserID != f.user {
+			t.Fatal("replay changed identity")
+		}
+		if err := unibox.CreateEntry(ctx, e.UserID, e.Message); err != nil {
+			return err
+		}
+		calls++
+		if calls == 1 {
+			return lost
+		}
+		return nil
+	}
+	if err := r.DeliverArrivals(ctx, handle); !errors.Is(err, lost) {
+		t.Fatalf("expected retained delivery: %v", err)
+	}
+	dueArrival(t, d, f)
+	r = NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d))
+	if err := r.DeliverArrivals(ctx, handle); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.DeliverArrivals(ctx, handle); err != nil {
+		t.Fatal(err)
+	}
+	var count int
+	if err := d.QueryRow(ctx, `SELECT count(*) FROM unibox_emails WHERE email_id=$1 AND id=$2`, f.mailbox, original).Scan(&count); err != nil || count != 1 || calls != 2 {
+		t.Fatalf("count=%d calls=%d err=%v", count, calls, err)
+	}
+	if pending, err := r.HasPendingArrival(ctx, f.user, f.mailbox, original); err != nil || pending {
+		t.Fatalf("pending not retired: %v", err)
+	}
+}
+
+func TestLiveSyncArrivalReportStagesOwnershipAndLegacyMaps(t *testing.T) {
+	d := liveUniboxFolderDB(t)
+	f := newUniboxFolderFixture(t, d.Pool)
+	ctx := context.Background()
+	r := NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d))
+	data, p := arrivalLivePayload(f, "reports")
+	p.Bounce = &models.JobEventInboundBounce{UserID: f.user, EmailID: f.mailbox, OriginalMessageID: "<send@test.local>"}
+	p.Complaint = &models.JobEventInboundComplaint{UserID: f.user, EmailID: f.mailbox, OriginalMessageID: "<send@test.local>"}
+	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")
+	}
+}