feat: prioritize durable mailbox arrivals that block dependent sync events and send-result processing

internal/app/consumer/arrival_outbox_test.go

@@ -3,14 +3,99 @@ package jobs
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
+type priorityArrivalStub struct {
+	repository.ArrivalOutbox
+	user, email, id uuid.UUID
+	deliver         func(context.Context, func(context.Context, models.JobEventType, any) error) (bool, error)
+}
+
+func (s *priorityArrivalStub) DeliverPendingArrival(ctx context.Context, user, email, id uuid.UUID, deliver func(context.Context, models.JobEventType, any) error) (bool, error) {
+	s.user, s.email, s.id = user, email, id
+	return s.deliver(ctx, deliver)
+}
+
+func TestDurableArrivalUnblocksDependentEventBeforeAcknowledgement(t *testing.T) {
+	for _, outcome := range []string{"delivered", "leased", "failed", "finished concurrently"} {
+		t.Run(outcome, func(t *testing.T) {
+			user, email, id := uuid.New(), uuid.New(), uuid.New()
+			pending := &syncArrivalPendingError{user: user, email: email, id: id}
+			ready, dependentCalls, arrivalCalls := false, 0, 0
+			failed := errors.New("arrival delivery failed")
+			outbox := &priorityArrivalStub{}
+			s := &JobsService{ArrivalOutbox: outbox}
+			s.InitEvents()
+			Register(s, models.JobEventTypeNewEmail, func(context.Context, *models.JobEventNewEmail) error {
+				arrivalCalls++
+				ready = true
+				return nil
+			})
+			Register(s, models.JobEventTypeRemoveEmail, func(context.Context, *models.JobEventRemoveEmail) error {
+				dependentCalls++
+				if !ready {
+					return fmt.Errorf("dependent event: %w", pending)
+				}
+				return nil
+			})
+			outbox.deliver = func(ctx context.Context, deliver func(context.Context, models.JobEventType, any) error) (bool, error) {
+				switch outcome {
+				case "leased":
+					return false, nil
+				case "failed":
+					return true, failed
+				case "finished concurrently":
+					ready = true
+					return false, nil
+				default:
+					return true, deliver(ctx, models.JobEventTypeNewEmail, &models.JobEventNewEmail{})
+				}
+			}
+			err := s.HandleEvent(t.Context(), &models.JobEvent{Type: models.JobEventTypeRemoveEmail, Body: &models.JobEventRemoveEmail{}})
+			if outbox.user != user || outbox.email != email || outbox.id != id {
+				t.Fatal("dependency lookup lost tenant/message scope")
+			}
+			switch outcome {
+			case "leased":
+				if !errors.Is(err, ErrSyncArrivalPending) || dependentCalls != 2 {
+					t.Fatalf("leased arrival acknowledged: calls=%d err=%v", dependentCalls, err)
+				}
+			case "failed":
+				if !errors.Is(err, failed) || dependentCalls != 1 {
+					t.Fatalf("failed arrival overtaken: calls=%d err=%v", dependentCalls, err)
+				}
+			default:
+				if err != nil || dependentCalls != 2 {
+					t.Fatalf("dependency did not unblock: calls=%d err=%v", dependentCalls, err)
+				}
+			}
+			if outcome == "delivered" && arrivalCalls != 1 {
+				t.Fatalf("arrival calls=%d", arrivalCalls)
+			}
+		})
+	}
+}
+
+func TestDurableArrivalDoesNotRetryOtherErrorsOrUnsupportedOutbox(t *testing.T) {
+	for _, err := range []error{errors.New("unrelated handler failure"), &syncArrivalPendingError{user: uuid.New(), email: uuid.New(), id: uuid.New()}} {
+		s := &JobsService{}
+		s.InitEvents()
+		calls := 0
+		Register(s, models.JobEventTypeRemoveEmail, func(context.Context, *models.JobEventRemoveEmail) error { calls++; return err })
+		if got := s.HandleEvent(t.Context(), &models.JobEvent{Type: models.JobEventTypeRemoveEmail, Body: &models.JobEventRemoveEmail{}}); !errors.Is(got, err) || calls != 1 {
+			t.Fatalf("unsupported recovery changed handler result: calls=%d err=%v", calls, got)
+		}
+	}
+}
+
 func TestDurableArrivalRequiresRegisteredHandler(t *testing.T) {
 	s := &JobsService{}
 	if err := s.deliverArrivalEvent(context.Background(), models.JobEventTypeNewEmail, &models.JobEventNewEmail{}); err == nil {

internal/app/consumer/event_remove_email.go

@@ -29,7 +29,7 @@ func (s *JobsService) HandleRemoveEmail(ctx context.Context, e *models.JobEventR
 			return err
 		}
 		if pending {
-			return ErrSyncArrivalPending
+			return &syncArrivalPendingError{user: e.UserID, email: e.EmailID, id: e.ID}
 		}
 	}
 	var checkErr error

internal/app/consumer/event_update_email.go

@@ -146,6 +146,13 @@ func (s *JobsService) applyRelayedFolder(ctx context.Context, e *models.JobEvent
 // emailForSyncUpdate rechecks visible mail if verification won the pending-row lock.
 var ErrSyncArrivalPending = errors.New("sync arrival pending delivery")
 
+type syncArrivalPendingError struct {
+	user, email, id uuid.UUID
+}
+
+func (e *syncArrivalPendingError) Error() string { return ErrSyncArrivalPending.Error() }
+func (e *syncArrivalPendingError) Unwrap() error { return ErrSyncArrivalPending }
+
 func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, emailID, id uuid.UUID, updatePending func(*models.EmailMessageStoreData)) (*models.EmailMessageStoreData, error) {
 	message, err := s.UniboxRepository.GetForSync(ctx, userID, emailID, id)
 	if !errors.Is(err, repository.ErrEmailNotFound) {
@@ -166,7 +173,7 @@ func (s *JobsService) emailForSyncUpdate(ctx context.Context, userID, emailID, i
 			return nil, lookupErr
 		}
 		if pending {
-			return nil, ErrSyncArrivalPending
+			return nil, &syncArrivalPendingError{user: userID, email: emailID, id: id}
 		}
 		// Delivery may have finished between the visible-row read and the marker read.
 		message, err = s.UniboxRepository.GetForSync(ctx, userID, emailID, id)

internal/app/consumer/events.go

@@ -3,10 +3,12 @@ package jobs
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"github.com/rs/zerolog/log"
 	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
 )
 
 type EventHandler[T any] func(ctx context.Context, event T) error
@@ -21,6 +23,19 @@ func (s *JobsService) HandleEvent(ctx context.Context, event *models.JobEvent) e
 			Msg("no handler registered for job event type, dropping")
 		return nil
 	}
+	err := resp(ctx, event.Body)
+	var pending *syncArrivalPendingError
+	if !errors.As(err, &pending) {
+		return err
+	}
+	outbox, ok := s.ArrivalOutbox.(repository.PriorityArrivalOutbox)
+	if !ok {
+		return err
+	}
+	// Resolve the blocking dependency without overtaking it or waiting for the whole backlog.
+	if _, deliveryErr := outbox.DeliverPendingArrival(ctx, pending.user, pending.email, pending.id, s.deliverArrivalEvent); deliveryErr != nil {
+		return deliveryErr
+	}
 	return resp(ctx, event.Body)
 }
 

internal/app/consumer/sync_truth_live_test.go

@@ -2,6 +2,7 @@ package jobs
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"testing"
 
@@ -14,6 +15,74 @@ import (
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
+func TestLiveSyncTruthDependentEventsPrioritizeArrival(t *testing.T) {
+	for _, kind := range []models.JobEventType{models.JobEventTypeEmailUpdate, models.JobEventTypeFolderUpdate, models.JobEventTypeRemoveEmail} {
+		t.Run(string(kind), func(t *testing.T) {
+			s, d := liveWarmupService(t)
+			f := newWarmupFixture(t, d)
+			key, err := kms.NewLocal(make([]byte, 32))
+			if err != nil {
+				t.Fatal(err)
+			}
+			r := repository.NewDurableEmailMessageMapRepository(d, cipher.NewService(key, nil, encryptedkeys.NewPostgres(d)))
+			s.ArrivalOutbox = r
+			s.InitEvents()
+			for i := 0; i < 41; i++ {
+				e := f.arrival("<"+uuid.NewString()+"@test.local>", nil)
+				e.Message.Subject = "ordinary customer question"
+				mapping := repository.EmailMessageData{UserID: e.UserID.String(), EmailID: e.Message.EmailID.String(), ID: e.Message.ID.String(), MessageID: e.Message.MessageID, ThreadID: e.Message.ThreadID}
+				if err := r.AdmitArrival(t.Context(), mapping, &repository.PendingArrival{Arrival: e}); err != nil {
+					t.Fatal(err)
+				}
+				if i != 40 {
+					continue
+				}
+				var body any
+				switch kind {
+				case models.JobEventTypeEmailUpdate:
+					body = &models.JobEventEmailUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Flags: []string{"\\Seen"}, UID: 99, Mailbox: 2, ModSeq: 5}
+				case models.JobEventTypeFolderUpdate:
+					body = &models.JobEventFolderUpdate{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID, Folder: models.FolderArchive}
+				case models.JobEventTypeRemoveEmail:
+					body = &models.JobEventRemoveEmail{UserID: e.UserID, EmailID: e.Message.EmailID, ID: e.Message.ID}
+				}
+				// JSON-decoded bodies use the same registered handlers as typed Kafka events.
+				raw, err := json.Marshal(body)
+				if err != nil {
+					t.Fatal(err)
+				}
+				var decoded map[string]any
+				if err := json.Unmarshal(raw, &decoded); err != nil {
+					t.Fatal(err)
+				}
+				event := &models.JobEvent{Type: kind, Body: decoded}
+				if err := s.HandleEvent(t.Context(), event); err != nil {
+					t.Fatalf("dependent event blocked by unrelated backlog: %v", err)
+				}
+				if err := s.HandleEvent(t.Context(), event); err != nil {
+					t.Fatalf("dependent replay failed: %v", err)
+				}
+				message, err := s.UniboxRepository.GetForSync(t.Context(), e.UserID, e.Message.EmailID, e.Message.ID)
+				if kind == models.JobEventTypeRemoveEmail {
+					if !errors.Is(err, repository.ErrEmailNotFound) {
+						t.Fatalf("removal did not follow arrival: %v", err)
+					}
+				} else if err != nil {
+					t.Fatal(err)
+				} else if kind == models.JobEventTypeFolderUpdate && message.Folder != models.FolderArchive {
+					t.Fatal("folder change lost after arrival")
+				} else if kind == models.JobEventTypeEmailUpdate && (message.UID != 99 || !message.Seen) {
+					t.Fatal("flags/provider update lost after arrival")
+				}
+			}
+			var left int
+			if err := d.QueryRow(t.Context(), `SELECT count(*) FROM sync_arrival_outbox WHERE email_id=$1`, f.partner).Scan(&left); err != nil || left != 40 {
+				t.Fatalf("unrelated arrivals lost: %d %v", left, err)
+			}
+		})
+	}
+}
+
 func TestLiveSyncTruthRemovalWaitsForArrivalAndLostStageAcknowledgement(t *testing.T) {
 	for _, lostAck := range []bool{false, true} {
 		for _, kind := range []string{"ordinary", "confirmed warmup", "unconfirmed warmup"} {

internal/repository/email_message_map.go

@@ -42,6 +42,10 @@ type ArrivalOutbox interface {
 	DeliverArrivals(context.Context, func(context.Context, models.JobEventType, any) error) error
 }
 
+type PriorityArrivalOutbox interface {
+	DeliverPendingArrival(context.Context, uuid.UUID, uuid.UUID, uuid.UUID, func(context.Context, models.JobEventType, any) error) (bool, error)
+}
+
 type ArrivalBacklog struct {
 	Pending int
 	Oldest  *time.Time

internal/repository/pg_sync_arrival.go

@@ -118,7 +118,7 @@ func (r *pgEmailMessageMapRepository) HasPendingArrival(ctx context.Context, use
 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)
+		found, err := r.deliverArrival(ctx, nil, deliver)
 		if err != nil && first == nil {
 			first = err
 		}
@@ -129,20 +129,37 @@ func (r *pgEmailMessageMapRepository) DeliverArrivals(ctx context.Context, deliv
 	return first
 }
 
-func (r *pgEmailMessageMapRepository) deliverArrival(ctx context.Context, deliver func(context.Context, models.JobEventType, any) error) (bool, error) {
+type arrivalIdentity struct {
+	user, email, id uuid.UUID
+}
+
+func (r *pgEmailMessageMapRepository) DeliverPendingArrival(ctx context.Context, user, email, id uuid.UUID, deliver func(context.Context, models.JobEventType, any) error) (bool, error) {
+	if user == uuid.Nil || email == uuid.Nil || id == uuid.Nil {
+		return false, errors.New("invalid pending arrival identity")
+	}
+	return r.deliverArrival(ctx, &arrivalIdentity{user: user, email: email, id: id}, deliver)
+}
+
+func (r *pgEmailMessageMapRepository) deliverArrival(ctx context.Context, target *arrivalIdentity, 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
+	scope := ""
+	var args []any
+	if target != nil {
+		scope = " AND o.user_id=$1 AND o.email_id=$2 AND o.id=$3"
+		args = []any{target.user, target.email, target.id}
+	}
 	// 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())
+		WHERE o.retry_at<=now() AND (o.locked_until IS NULL OR o.locked_until<now())`+scope+`
 		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)
+		RETURNING o.user_id,o.email_id,o.organization_id,o.message_id,o.payload,o.stage,o.lease`, args...).Scan(&user, &email, &org, &key, &payload, &stage, &lease)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return false, nil
 	}

internal/repository/pg_sync_arrival_live_test.go

@@ -38,6 +38,69 @@ func dueArrival(t *testing.T, d *db.DB, f *uniboxFolderFixture) {
 	}
 }
 
+func TestLiveSyncArrivalPriorityRetainsScopeLeaseRetryAndBacklog(t *testing.T) {
+	d := liveUniboxFolderDB(t)
+	f := newUniboxFolderFixture(t, d.Pool)
+	r := NewDurableEmailMessageMapRepository(d, arrivalLiveCipher(t, d))
+	ctx := t.Context()
+	for i := 0; i < 40; i++ {
+		data, pending := arrivalLivePayload(f, uuid.NewString())
+		if err := r.AdmitArrival(ctx, data, pending); err != nil {
+			t.Fatal(err)
+		}
+	}
+	data, pending := arrivalLivePayload(f, "blocking-arrival")
+	if err := r.AdmitArrival(ctx, data, pending); err != nil {
+		t.Fatal(err)
+	}
+	id := pending.Arrival.Message.ID
+	deliveries := 0
+	deliver := func(_ context.Context, kind models.JobEventType, body any) error {
+		if kind != models.JobEventTypeNewEmail || body.(*models.JobEventNewEmail).Message.ID != id {
+			t.Fatal("priority delivery replayed an unrelated arrival")
+		}
+		deliveries++
+		return nil
+	}
+	other := newUniboxFolderFixture(t, d.Pool)
+	for _, scope := range [][3]uuid.UUID{{other.user, f.mailbox, id}, {f.user, other.mailbox, id}, {f.user, f.mailbox, uuid.Nil}} {
+		found, err := r.DeliverPendingArrival(ctx, scope[0], scope[1], scope[2], deliver)
+		if found || (scope[2] == uuid.Nil && err == nil) || deliveries != 0 {
+			t.Fatalf("invalid ownership or identity delivered: %v %v", found, err)
+		}
+	}
+	for _, column := range []string{"locked_until", "retry_at"} {
+		if _, err := d.Exec(ctx, `UPDATE sync_arrival_outbox SET `+column+`=now()+interval '1 minute' WHERE id=$1`, id); err != nil {
+			t.Fatal(err)
+		}
+		if found, err := r.DeliverPendingArrival(ctx, f.user, f.mailbox, id, deliver); found || err != nil || deliveries != 0 {
+			t.Fatalf("priority bypassed %s: %v %v", column, found, err)
+		}
+		dueArrival(t, d, f)
+	}
+	failed := errors.New("priority handler failed")
+	if found, err := r.DeliverPendingArrival(ctx, f.user, f.mailbox, id, func(context.Context, models.JobEventType, any) error { return failed }); !found || !errors.Is(err, failed) {
+		t.Fatalf("failed priority delivery was acknowledged: %v %v", found, err)
+	}
+	if retained, err := r.HasPendingArrival(ctx, f.user, f.mailbox, id); !retained || err != nil {
+		t.Fatalf("failed stage discarded: %v %v", retained, err)
+	}
+	if found, err := r.DeliverPendingArrival(ctx, f.user, f.mailbox, id, deliver); found || err != nil {
+		t.Fatalf("priority bypassed failure retry delay: %v %v", found, err)
+	}
+	dueArrival(t, d, f)
+	if found, err := r.DeliverPendingArrival(ctx, f.user, f.mailbox, id, deliver); !found || err != nil || deliveries != 1 {
+		t.Fatalf("blocking dependency not prioritized: %v %v calls=%d", found, err, deliveries)
+	}
+	var left int
+	if err := d.QueryRow(ctx, `SELECT count(*) FROM sync_arrival_outbox WHERE email_id=$1`, f.mailbox).Scan(&left); err != nil || left != 40 {
+		t.Fatalf("priority delivery lost the unrelated backlog: %d %v", left, err)
+	}
+	if found, err := r.DeliverPendingArrival(ctx, f.user, f.mailbox, id, deliver); found || err != nil || deliveries != 1 {
+		t.Fatalf("completed arrival redelivered: %v %v", found, err)
+	}
+}
+
 func TestLiveSyncArrivalRestartAndAmbiguousAdmission(t *testing.T) {
 	d := liveUniboxFolderDB(t)
 	f := newUniboxFolderFixture(t, d.Pool)