diff --git a/internal/app/consumer/arrival_outbox_test.go b/internal/app/consumer/arrival_outbox_test.go index 857ffaa8b..7e0ad4369 100644 --- a/internal/app/consumer/arrival_outbox_test.go +++ b/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 { diff --git a/internal/app/consumer/event_remove_email.go b/internal/app/consumer/event_remove_email.go index 0bf1a9509..31d74c4e9 100644 --- a/internal/app/consumer/event_remove_email.go +++ b/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 diff --git a/internal/app/consumer/event_update_email.go b/internal/app/consumer/event_update_email.go index 216995de3..f5ec9989d 100644 --- a/internal/app/consumer/event_update_email.go +++ b/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) diff --git a/internal/app/consumer/events.go b/internal/app/consumer/events.go index 5c91ebf28..417ba9849 100644 --- a/internal/app/consumer/events.go +++ b/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) } diff --git a/internal/app/consumer/sync_truth_live_test.go b/internal/app/consumer/sync_truth_live_test.go index 1105e090c..b2c686f18 100644 --- a/internal/app/consumer/sync_truth_live_test.go +++ b/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"} { diff --git a/internal/repository/email_message_map.go b/internal/repository/email_message_map.go index 2d2de8591..f2c4b530d 100644 --- a/internal/repository/email_message_map.go +++ b/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 diff --git a/internal/repository/pg_sync_arrival.go b/internal/repository/pg_sync_arrival.go index f67d6b34f..3fe29d35b 100644 --- a/internal/repository/pg_sync_arrival.go +++ b/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