diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0e3ea3289..9902bcfa5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -261,12 +261,12 @@ jobs: go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveMailTransport(CLIStartup|UpgradePreservesHistoryAndRollback)$' -count=1 | tee mailbox-transport-upgrade.log grep -q '^--- PASS: TestLiveMailTransportCLIStartup ' mailbox-transport-upgrade.log grep -q '^--- PASS: TestLiveMailTransportUpgradePreservesHistoryAndRollback ' mailbox-transport-upgrade.log - ! grep -q -- '--- SKIP:' mailbox-transport-upgrade.log - go test -race -v -p 1 ./internal/repository ./internal/app/orgtransfer ./internal/app/consumer -run '^TestLive(MailTransport.*|ImportMailTransport.*)$' -count=1 | tee mailbox-transport-runtime.log - for test in TestLiveMailTransportNativeDomainExcludesHistory TestLiveMailTransportImportSerializesWithConfirmedResult TestLiveImportMailTransportCheckpointCountsAndUnchangedRows TestLiveMailTransportConfirmedSendReconcilesOnlySend TestLiveImportMailTransportLateAndSpanningSuccess TestLiveImportMailTransportCredentialOnlyWritesRotateFence TestLiveImportMailTransportExportOmitsGeneration TestLiveImportMailTransportWholeArchiveOmitsGeneration TestLiveMailTransportSnapshotKeepsConfigurationAndGenerationTogether TestLiveMailTransportOperatorConfirmationDoesNotRecover; do + if grep -q -- '--- SKIP:' mailbox-transport-upgrade.log; then exit 1; fi + go test -race -v -p 1 ./internal/repository ./internal/app/orgtransfer ./internal/app/consumer ./internal/app/email -run '^TestLive(MailTransport.*|ImportMailTransport.*|WorkerMailboxLoadReadsOneGenerationWithSingleConnection)$' -count=1 | tee mailbox-transport-runtime.log + for test in TestLiveMailTransportNativeDomainExcludesHistory TestLiveMailTransportImportSerializesWithConfirmedResult TestLiveImportMailTransportCheckpointCountsAndUnchangedRows TestLiveMailTransportConfirmedSendReconcilesOnlySend TestLiveImportMailTransportLateAndSpanningSuccess TestLiveImportMailTransportCredentialOnlyWritesRotateFence TestLiveImportMailTransportExportOmitsGeneration TestLiveImportMailTransportWholeArchiveOmitsGeneration TestLiveMailTransportSnapshotKeepsConfigurationAndGenerationTogether TestLiveWorkerMailboxLoadReadsOneGenerationWithSingleConnection TestLiveMailTransportOperatorConfirmationDoesNotRecover; do grep -q "^--- PASS: $test " mailbox-transport-runtime.log done - ! grep -q -- '--- SKIP:' mailbox-transport-runtime.log + if grep -q -- '--- SKIP:' mailbox-transport-runtime.log; then exit 1; fi grep -q '^--- PASS: TestLiveImportMailTransportWholeArchiveSendResultLockOrder ' mailbox-transport-runtime.log for logfile in mailbox-transport-upgrade.log mailbox-transport-runtime.log; do test "$(grep -c '^=== RUN' "$logfile")" -gt 0 @@ -275,7 +275,7 @@ jobs: go test -race -v -p 1 ./internal/app/worker/wmail ./internal/app/worker/mailmanager ./internal/app/consumer -run '^Test(MailTransportGeneration.*|SyncState.*)$' -count=1 | tee mailbox-transport-worker.log grep -q '^--- PASS: TestMailTransportGenerationRequiresFreshCompletedSyncPass ' mailbox-transport-worker.log grep -q '^--- PASS: TestMailTransportGenerationReplacesUnchangedCredentialsAndDrains ' mailbox-transport-worker.log - ! grep -q -- '--- SKIP:' mailbox-transport-worker.log + if grep -q -- '--- SKIP:' mailbox-transport-worker.log; then exit 1; fi - name: Typed mailbox JSON and native Avro registry compatibility env: @@ -296,7 +296,7 @@ jobs: for testcase in add send sync error; do grep -q -- "--- PASS: TestMailTransportCodecRegistryCompatibility/$testcase " mailbox-transport-codec.log done - ! grep -q -- '--- SKIP:' mailbox-transport-codec.log + if grep -q -- '--- SKIP:' mailbox-transport-codec.log; then exit 1; fi printf 'codec RUN=%s PASS=%s SKIP=0\n' "$(grep -c '^=== RUN' mailbox-transport-codec.log)" "$(grep -c -- '--- PASS:' mailbox-transport-codec.log)" - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL) diff --git a/internal/app/consumer/event_sync_state.go b/internal/app/consumer/event_sync_state.go index 12bb0ac0c..dbb132329 100644 --- a/internal/app/consumer/event_sync_state.go +++ b/internal/app/consumer/event_sync_state.go @@ -28,7 +28,7 @@ func (s *JobsService) HandleSyncState(ctx context.Context, e *models.JobEventSyn // Persist recovery before state, so a failed proof write remains retryable. if e.State.LastSyncedAt != nil && e.State.Deferred == 0 && e.State.BackfillCursor.GoogleRecovery == nil && - (prev == nil || prev.LastSyncedAt == nil || e.State.TransportGeneration != prev.TransportGeneration || !models.ValidMailTransportTime(*prev.LastSyncedAt, time.Now()) || e.State.LastSyncedAt.After(*prev.LastSyncedAt)) { + (prev == nil || prev.LastSyncedAt == nil || !models.ValidMailTransportTime(*prev.LastSyncedAt, time.Now()) || e.State.LastSyncedAt.After(*prev.LastSyncedAt)) { if err := s.resolveTransientMailErrors(ctx, e.EmailID, e.State.LastSyncedAt, e.State.TransportGeneration); err != nil { return err } diff --git a/internal/app/consumer/sync_truth_test.go b/internal/app/consumer/sync_truth_test.go index 761bb411d..1197f84b2 100644 --- a/internal/app/consumer/sync_truth_test.go +++ b/internal/app/consumer/sync_truth_test.go @@ -46,23 +46,25 @@ func TestSyncTruthProgressRelaysDoNotResolveTransportWarnings(t *testing.T) { previous := time.Now().Add(-time.Hour) fresh := previous.Add(time.Minute) for _, tc := range []struct { - name string - stamp *time.Time - deferred int - recovery *models.GoogleHistoryRecovery - resolved int + name string + stamp *time.Time + generation uuid.UUID + deferred int + recovery *models.GoogleHistoryRecovery + resolved int }{ - {"heartbeat", &previous, 0, nil, 0}, - {"older redelivery", nil, 0, nil, 0}, - {"deferred", &fresh, 5, nil, 0}, - {"recovering", &fresh, 0, &models.GoogleHistoryRecovery{HistoryID: 100}, 0}, - {"successful catch-up", &fresh, 0, nil, 1}, + {"heartbeat", &previous, syncTestGeneration, 0, nil, 0}, + {"older redelivery", nil, syncTestGeneration, 0, nil, 0}, + {"deferred", &fresh, syncTestGeneration, 5, nil, 0}, + {"recovering", &fresh, syncTestGeneration, 0, &models.GoogleHistoryRecovery{HistoryID: 100}, 0}, + {"unproven catch-up", &fresh, uuid.Nil, 0, nil, 0}, + {"successful catch-up", &fresh, syncTestGeneration, 0, nil, 1}, } { t.Run(tc.name, func(t *testing.T) { errRepo := &stubErrorRepo{} repo := &stubSyncStateRepo{saved: &models.SyncState{LastSyncedAt: &previous}} s := &JobsService{EmailSyncStateRepository: repo, EmailAccountErrorRepository: errRepo} - if err := s.HandleSyncState(t.Context(), &models.JobEventSyncState{UserID: uuid.New(), EmailID: uuid.New(), State: models.SyncState{BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: tc.stamp, Deferred: tc.deferred, BackfillCursor: models.SyncCursor{GoogleRecovery: tc.recovery}}}); err != nil { + if err := s.HandleSyncState(t.Context(), &models.JobEventSyncState{UserID: uuid.New(), EmailID: uuid.New(), State: models.SyncState{TransportGeneration: tc.generation, BackfillStatus: models.SyncBackfillComplete, LastSyncedAt: tc.stamp, Deferred: tc.deferred, BackfillCursor: models.SyncCursor{GoogleRecovery: tc.recovery}}}); err != nil { t.Fatal(err) } if repo.put == nil || errRepo.calls != tc.resolved { diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index 83e2949fd..4cbe3b713 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -323,14 +323,51 @@ func (s *emailService) releaseDeadWorker(ctx context.Context, accountID uuid.UUI return nil, nil } -// buildAddWorkerEmail reconstructs the worker payload for an account, decrypting -// its credentials and attaching the provider-specific data. Cfg is intentionally -// left zero: it is avro-excluded and the worker rebuilds it locally from its own -// oauth config. +type addWorkerEmailPreload struct { + syncData *models.AddWorkerEmailSyncData + historyID uint64 + delta map[string]string + mailboxes []models.Mailbox + managed bool +} + +func (s *emailService) preloadAddWorkerEmail(ctx context.Context, acc *models.Email) (*addWorkerEmailPreload, error) { + userID, err := uuid.Parse(acc.UserID) + if err != nil { + return nil, err + } + syncData, err := s.syncDataFor(ctx, acc.ID, false) + if err != nil { + return nil, err + } + p := &addWorkerEmailPreload{syncData: syncData} + switch models.InboxProvider(acc.Provider) { + case models.InboxProviderGoogle: + p.historyID = s.lastHistoryFor(ctx, userID, acc.ID, nil) + case models.InboxProviderOutlook: + p.delta, err = s.deltaLinksFor(ctx, userID, acc.ID) + if err != nil { + return nil, err + } + case models.InboxProviderSMTPIMAP: + p.mailboxes = s.mailboxesFor(ctx, userID, acc.ID) + } + if acc.AuthMethod != models.MailAuthDelegated && s.cloudLink != nil { + if link, err := s.cloudLink.GetByAccount(ctx, acc.ID); err == nil && link != nil { + p.managed = link.Managed + } + } + return p, nil +} + func (s *emailService) buildAddWorkerEmail(ctx context.Context, acc *models.Email) (*models.AddWorkerEmail, error) { if snapshots, ok := s.emailRepository.(repository.EmailTransportSnapshot); ok { + preload, err := s.preloadAddWorkerEmail(ctx, acc) + if err != nil { + return nil, err + } var payload *models.AddWorkerEmail - err := snapshots.WithTransportSnapshot(ctx, func(repo repository.EmailRepository) error { + err = snapshots.WithTransportSnapshot(ctx, func(repo repository.EmailRepository) error { current, xerr := repo.GetByID(ctx, acc.ID) if xerr != nil { return xerr @@ -338,16 +375,19 @@ func (s *emailService) buildAddWorkerEmail(ctx context.Context, acc *models.Emai if current == nil || current.Status != "active" { return nil } + if current.TransportGeneration != acc.TransportGeneration || current.Provider != acc.Provider || current.UserID != acc.UserID || current.AuthMethod != acc.AuthMethod { + return fmt.Errorf("mailbox changed while preparing worker load") + } var err error - payload, err = s.buildAddWorkerEmailSnapshot(ctx, current, repo) + payload, err = s.buildAddWorkerEmailSnapshot(ctx, current, repo, preload) return err }) return payload, err } - return s.buildAddWorkerEmailSnapshot(ctx, acc, s.emailRepository) + return s.buildAddWorkerEmailSnapshot(ctx, acc, s.emailRepository, nil) } -func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *models.Email, repo repository.EmailRepository) (*models.AddWorkerEmail, error) { +func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *models.Email, repo repository.EmailRepository, preload *addWorkerEmailPreload) (*models.AddWorkerEmail, error) { userID, err := uuid.Parse(acc.UserID) if err != nil { return nil, err @@ -356,9 +396,21 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod provider := models.InboxProvider(acc.Provider) saveToSent := acc.SaveToSent - sync, err := s.syncDataFor(ctx, acc.ID) - if err != nil { - return nil, err + var sync *models.AddWorkerEmailSyncData + if preload != nil { + folders, xerr := repo.GetSyncSkipFolders(ctx, acc.ID) + if xerr != nil { + return nil, fmt.Errorf("sync skip folders lookup: %w", xerr) + } + copy := *preload.syncData + copy.Policy.SkipFolders = folders + sync = © + } else { + var err error + sync, err = s.syncDataFor(ctx, acc.ID, true) + if err != nil { + return nil, err + } } out := &models.AddWorkerEmail{ TransportGeneration: acc.TransportGeneration, @@ -374,9 +426,15 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod SaveToSent: &saveToSent, } if provider == models.InboxProviderOutlook { - links, err := s.deltaLinksFor(ctx, userID, acc.ID) - if err != nil { - return nil, err + var links map[string]string + if preload != nil { + links = preload.delta + } else { + var err error + links, err = s.deltaLinksFor(ctx, userID, acc.ID) + if err != nil { + return nil, err + } } out.Graph = &models.AddWorkerEmailGraphData{DeltaLinks: links} } @@ -394,7 +452,7 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod out.Brokered = true switch provider { case models.InboxProviderGoogle: - out.Google = &models.AddWorkerEmailGoogleData{LastHistoryID: s.lastHistoryFor(ctx, userID, acc.ID, acc.LastID)} + out.Google = &models.AddWorkerEmailGoogleData{LastHistoryID: s.workerHistoryID(ctx, userID, acc, preload)} case models.InboxProviderOutlook: out.Graph.User = d.Subject default: @@ -405,11 +463,16 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod // A managed mailbox has no local credential; the worker draws brokered tokens. if s.cloudLink != nil { - if m, err := s.cloudLink.GetByAccount(ctx, acc.ID); err == nil && m != nil && m.Managed { + managed := preload != nil && preload.managed + if preload == nil { + m, err := s.cloudLink.GetByAccount(ctx, acc.ID) + managed = err == nil && m != nil && m.Managed + } + if managed { out.Brokered = true switch provider { case models.InboxProviderGoogle: - out.Google = &models.AddWorkerEmailGoogleData{LastHistoryID: s.lastHistoryFor(ctx, userID, acc.ID, acc.LastID)} + out.Google = &models.AddWorkerEmailGoogleData{LastHistoryID: s.workerHistoryID(ctx, userID, acc, preload)} case models.InboxProviderOutlook: // The checkpoint was loaded before selecting the token owner. default: @@ -427,7 +490,7 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod } out.Google = &models.AddWorkerEmailGoogleData{ Token: oauthToken(creds), - LastHistoryID: s.lastHistoryFor(ctx, userID, acc.ID, acc.LastID), + LastHistoryID: s.workerHistoryID(ctx, userID, acc, preload), } case models.InboxProviderOutlook: creds, cerr := repo.GetOAuthCredentials(ctx, acc.ID) @@ -446,7 +509,7 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod SMTP: &models.Service{Host: creds.SMTPHost, Port: creds.SMTPPort, Username: creds.SMTPUser, Password: creds.SMTPPassword, Security: creds.SMTPSecurity}, IMAP: &models.Service{Host: creds.IMAPHost, Port: creds.IMAPPort, Username: creds.IMAPUser, Password: creds.IMAPPassword, Security: creds.IMAPSecurity}, }, - Mailboxes: s.mailboxesFor(ctx, userID, acc.ID), + Mailboxes: s.workerMailboxes(ctx, userID, acc, preload), } default: return nil, nil @@ -455,6 +518,26 @@ func (s *emailService) buildAddWorkerEmailSnapshot(ctx context.Context, acc *mod return out, nil } +func (s *emailService) workerHistoryID(ctx context.Context, userID uuid.UUID, acc *models.Email, preload *addWorkerEmailPreload) uint64 { + if preload == nil { + return s.lastHistoryFor(ctx, userID, acc.ID, acc.LastID) + } + if preload.historyID > 0 { + return preload.historyID + } + if acc.LastID != nil && *acc.LastID > 0 { + return uint64(*acc.LastID) + } + return 0 +} + +func (s *emailService) workerMailboxes(ctx context.Context, userID uuid.UUID, acc *models.Email, preload *addWorkerEmailPreload) []models.Mailbox { + if preload != nil { + return preload.mailboxes + } + return s.mailboxesFor(ctx, userID, acc.ID) +} + // lastHistoryFor is the Gmail checkpoint a (re)loaded mailbox resumes from. // // The consumer writes every checkpoint to email_history_ids. This used to read @@ -482,7 +565,7 @@ func (s *emailService) lastHistoryFor(ctx context.Context, userID, emailID uuid. // state a previous worker left behind. Policy comes from instance settings // (compiled defaults when none are wired), so an operator's change applies at // the next load: onboarding, reassignment, or the reconciler's republish. -func (s *emailService) syncDataFor(ctx context.Context, emailID uuid.UUID) (*models.AddWorkerEmailSyncData, error) { +func (s *emailService) syncDataFor(ctx context.Context, emailID uuid.UUID, readFolders bool) (*models.AddWorkerEmailSyncData, error) { budget := instancesettings.DefaultSync() if s.syncBudget != nil { budget = s.syncBudget.SyncBudget(ctx) @@ -499,11 +582,13 @@ func (s *emailService) syncDataFor(ctx context.Context, emailID uuid.UUID) (*mod // mailbox, so a failed read cannot fall back to "skip nothing": that // would have the worker baseline and import the excluded folders until // the next republish. The load fails instead and the reconciler retries. - skip, xerr := s.emailRepository.GetSyncSkipFolders(ctx, emailID) - if xerr != nil { - return nil, fmt.Errorf("sync skip folders lookup: %w", xerr) + if readFolders { + skip, xerr := s.emailRepository.GetSyncSkipFolders(ctx, emailID) + if xerr != nil { + return nil, fmt.Errorf("sync skip folders lookup: %w", xerr) + } + data.Policy.SkipFolders = skip } - data.Policy.SkipFolders = skip // A pool-linked mailbox is a warmup-only mirror: no history import. if s.poolLink != nil { if linked, err := s.poolLink.GetMailboxByAccount(ctx, emailID); err == nil && linked != nil { diff --git a/internal/app/email/mail_transport_snapshot_live_test.go b/internal/app/email/mail_transport_snapshot_live_test.go new file mode 100644 index 000000000..f4b25fa0a --- /dev/null +++ b/internal/app/email/mail_transport_snapshot_live_test.go @@ -0,0 +1,76 @@ +package email + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/infrastructure/db" + "github.com/warmbly/warmbly/internal/pkg/encrypt" + "github.com/warmbly/warmbly/internal/repository" +) + +func TestLiveWorkerMailboxLoadReadsOneGenerationWithSingleConnection(t *testing.T) { + f := newRemovalLiveFixture(t) + if _, err := f.pool.Exec(t.Context(), `UPDATE email_accounts SET provider='gmail',sync_skip_folders=ARRAY['Old'] WHERE id=$1`, f.mailbox); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(t.Context(), `INSERT INTO email_accounts_oauth(email_account_id,access_token,refresh_token,expires_at) VALUES($1,'old-access','old-refresh',NOW()+INTERVAL '1 hour')`, f.mailbox); err != nil { + t.Fatal(err) + } + config := f.pool.Config().Copy() + config.MaxConns = 1 + config.MinConns = 0 + limited, err := pgxpool.NewWithConfig(t.Context(), config) + if err != nil { + t.Fatal(err) + } + defer limited.Close() + enc, err := encrypt.NewEncrypter(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + handle := &db.DB{Pool: limited} + repo := repository.NewEmailRepostory(handle, enc) + s := &emailService{ + emailRepository: repo, + syncState: repository.NewEmailSyncStateRepository(handle), + historyID: repository.NewEmailHistoryIDRepository(handle), + } + before, xerr := repo.GetByID(t.Context(), f.mailbox) + if xerr != nil { + t.Fatal(xerr) + } + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + loaded, err := s.buildAddWorkerEmail(ctx, before) + if err != nil { + t.Fatal(err) + } + if loaded.TransportGeneration != before.TransportGeneration || loaded.Google == nil || loaded.Google.Token.AccessToken != "old-access" || len(loaded.Sync.Policy.SkipFolders) != 1 || loaded.Sync.Policy.SkipFolders[0] != "Old" { + t.Fatal("single-connection load lost its credential or policy snapshot") + } + next := uuid.New() + if _, err := f.pool.Exec(ctx, `UPDATE email_accounts SET transport_generation=$2,sync_skip_folders=ARRAY['Fresh'] WHERE id=$1`, f.mailbox, next); err != nil { + t.Fatal(err) + } + if _, err := f.pool.Exec(ctx, `UPDATE email_accounts_oauth SET access_token='fresh-access' WHERE email_account_id=$1`, f.mailbox); err != nil { + t.Fatal(err) + } + if loaded, err := s.buildAddWorkerEmail(ctx, before); err == nil || loaded != nil { + t.Fatal("outdated preload crossed the restore fence") + } + after, xerr := repo.GetByID(ctx, f.mailbox) + if xerr != nil { + t.Fatal(xerr) + } + loaded, err = s.buildAddWorkerEmail(ctx, after) + if err != nil { + t.Fatal(err) + } + if loaded.TransportGeneration != next || loaded.Google.Token.AccessToken != "fresh-access" || len(loaded.Sync.Policy.SkipFolders) != 1 || loaded.Sync.Policy.SkipFolders[0] != "Fresh" { + t.Fatal("fresh load mixed old and new configuration") + } +} diff --git a/internal/app/email/service.go b/internal/app/email/service.go index aa70eb618..7b9555777 100644 --- a/internal/app/email/service.go +++ b/internal/app/email/service.go @@ -422,7 +422,7 @@ func (s *emailService) GetSyncState(ctx context.Context, orgID, emailID string) if xerr != nil { return nil, models.SyncPolicy{}, nil, xerr } - data, err := s.syncDataFor(ctx, acc.ID) + data, err := s.syncDataFor(ctx, acc.ID, true) if err != nil { log.Error().Err(err).Str("email_id", acc.ID.String()).Msg("sync state: policy lookup failed") return nil, models.SyncPolicy{}, nil, errx.InternalError() diff --git a/internal/app/eventschemas/testdata/jobs.worker-events.avsc b/internal/app/eventschemas/testdata/jobs.worker-events.avsc index eefcbcec1..6ac84a4b5 100644 --- a/internal/app/eventschemas/testdata/jobs.worker-events.avsc +++ b/internal/app/eventschemas/testdata/jobs.worker-events.avsc @@ -9,6 +9,16 @@ "type": [ { "fields": [ + { + "default": "", + "name": "transport_generation", + "type": "string" + }, + { + "default": "", + "name": "operation", + "type": "string" + }, { "default": "", "name": "task_id", @@ -709,11 +719,17 @@ "folders_skipped_conflict": 0, "last_synced_at": null, "throttle_reason": "", - "throttled_until": null + "throttled_until": null, + "transport_generation": "" }, "name": "state", "type": { "fields": [ + { + "default": "", + "name": "transport_generation", + "type": "string" + }, { "default": "", "name": "backfill_status", @@ -979,6 +995,11 @@ }, { "fields": [ + { + "default": "", + "name": "transport_generation", + "type": "string" + }, { "default": "", "name": "task_id", diff --git a/internal/app/eventschemas/testdata/worker-commands.avsc b/internal/app/eventschemas/testdata/worker-commands.avsc index 5080e9494..c7c3a06e7 100644 --- a/internal/app/eventschemas/testdata/worker-commands.avsc +++ b/internal/app/eventschemas/testdata/worker-commands.avsc @@ -9,6 +9,11 @@ "type": [ { "fields": [ + { + "default": "", + "name": "transport_generation", + "type": "string" + }, { "default": "", "name": "id", @@ -354,6 +359,11 @@ "null", { "fields": [ + { + "default": "", + "name": "transport_generation", + "type": "string" + }, { "default": "", "name": "backfill_status", diff --git a/internal/infrastructure/codec/mail_transport_registry_test.go b/internal/infrastructure/codec/mail_transport_registry_test.go index 15a856ce7..8cb7dabea 100644 --- a/internal/infrastructure/codec/mail_transport_registry_test.go +++ b/internal/infrastructure/codec/mail_transport_registry_test.go @@ -141,7 +141,7 @@ func TestMailTransportCodecRegistryCompatibility(t *testing.T) { } url := os.Getenv("SR_URL") if url == "" { - t.Fatal("SR_URL must name an isolated local Schema Registry") + t.Skip("SR_URL must name an isolated local Schema Registry") } generation, task, mailbox := uuid.New(), uuid.New(), uuid.New() at := time.Now().UTC().Truncate(time.Millisecond) diff --git a/internal/models/event_schema_test.go b/internal/models/event_schema_test.go index 8e01de4d5..61bc52776 100644 --- a/internal/models/event_schema_test.go +++ b/internal/models/event_schema_test.go @@ -2,6 +2,7 @@ package models import ( "encoding/base64" + "encoding/json" "fmt" "reflect" "testing" @@ -69,13 +70,16 @@ func TestEventAvroReadsPreUpgradeEnvelopes(t *testing.T) { t.Fatal(err) } var got WorkerEvent - if err := EventAvro.Unmarshal(WorkerEvent{}.Schema(), payload, &got); err != nil { + if err := EventAvro.Unmarshal(preTransportSchema(t, WorkerEvent{}.Schema()), payload, &got); err != nil { t.Fatal(err) } if got.Type != WorkerEventTypeAddEmail { t.Fatalf("worker type changed: %s", got.Type) } - assertBody(t, got.Body, sample(WorkerEventBodies[WorkerEventTypeAddEmail])) + want := sample(WorkerEventBodies[WorkerEventTypeAddEmail]).(*AddWorkerEmail) + want.TransportGeneration = [16]byte{} + want.Sync.State.TransportGeneration = [16]byte{} + assertBody(t, got.Body, want) }) t.Run("job", func(t *testing.T) { payload, err := base64.StdEncoding.DecodeString("FEVNQUlMX1NFTlQiSDgxODE4MTgxLTgxODEtODE4MS04MTgxLTgxODE4MTgxODE4MQECeAJ4AniA8IbC7mUCAgJ4AngOAngCeAJ4AngCeIDwhsLuZQKA8IbC7mUCeAJ4AngCeAECeAJ4AngCeAJ4") @@ -83,16 +87,68 @@ func TestEventAvroReadsPreUpgradeEnvelopes(t *testing.T) { t.Fatal(err) } var got JobEvent - if err := EventAvro.Unmarshal(JobEvent{}.Schema(), payload, &got); err != nil { + if err := EventAvro.Unmarshal(preTransportSchema(t, JobEvent{}.Schema()), payload, &got); err != nil { t.Fatal(err) } if got.Type != JobEventTypeEmailSent { t.Fatalf("job type changed: %s", got.Type) } - assertBody(t, got.Body, sample(JobEventBodies[JobEventTypeEmailSent])) + want := sample(JobEventBodies[JobEventTypeEmailSent]).(SendEmailResult) + want.TransportGeneration = [16]byte{} + assertBody(t, got.Body, want) }) } +// Unframed fixtures need their writer schema, just as registry decoding does. +func preTransportSchema(t *testing.T, schema avro.Schema) avro.Schema { + t.Helper() + doc, err := SchemaDocument(schema) + if err != nil { + t.Fatal(err) + } + var root any + if err := json.Unmarshal([]byte(doc), &root); err != nil { + t.Fatal(err) + } + var strip func(any) + strip = func(v any) { + switch x := v.(type) { + case map[string]any: + if fields, ok := x["fields"].([]any); ok { + kept := fields[:0] + for _, f := range fields { + field := f.(map[string]any) + if field["name"] == "transport_generation" || x["name"] == "EmailErrorEvent" && field["name"] == "operation" { + continue + } + strip(f) + kept = append(kept, f) + } + x["fields"] = kept + } + for k, value := range x { + if k != "fields" { + strip(value) + } + } + case []any: + for _, value := range x { + strip(value) + } + } + } + strip(root) + old, err := json.Marshal(root) + if err != nil { + t.Fatal(err) + } + parsed, err := avro.Parse(string(old)) + if err != nil { + t.Fatal(err) + } + return parsed +} + // The decoded body has to arrive as the type the handler asserts on, carrying // what was sent. Both halves matter and only the first was checked here for a // while: a body whose type is right and whose fields are empty is what a codec diff --git a/internal/repository/mail_transport_snapshot_live_test.go b/internal/repository/mail_transport_snapshot_live_test.go index e1b512f75..600d1cb7a 100644 --- a/internal/repository/mail_transport_snapshot_live_test.go +++ b/internal/repository/mail_transport_snapshot_live_test.go @@ -3,8 +3,11 @@ package repository import ( "context" "testing" + "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/infrastructure/db" "github.com/warmbly/warmbly/internal/pkg/encrypt" ) @@ -14,13 +17,26 @@ func TestLiveMailTransportSnapshotKeepsConfigurationAndGenerationTogether(t *tes if err != nil { t.Fatal(err) } - repo := NewEmailRepostory(f.db, enc) + config := f.pool.Config().Copy() + config.MaxConns = 1 + config.MinConns = 0 + limited, err := pgxpool.NewWithConfig(t.Context(), config) + if err != nil { + t.Fatal(err) + } + defer limited.Close() + repo := NewEmailRepostory(&db.DB{Pool: limited}, enc) + if _, err := f.pool.Exec(t.Context(), `UPDATE email_accounts SET sync_skip_folders=ARRAY['INBOX'] WHERE id=$1`, f.account); err != nil { + t.Fatal(err) + } if _, err := f.pool.Exec(t.Context(), `INSERT INTO email_accounts_oauth(email_account_id,access_token,refresh_token,expires_at) VALUES($1,'old-access','old-refresh',NOW()+INTERVAL '1 hour')`, f.account); err != nil { t.Fatal(err) } next := uuid.New() - if err := repo.(EmailTransportSnapshot).WithTransportSnapshot(t.Context(), func(snapshot EmailRepository) error { - account, xerr := snapshot.GetByID(t.Context(), f.account) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + if err := repo.(EmailTransportSnapshot).WithTransportSnapshot(ctx, func(snapshot EmailRepository) error { + account, xerr := snapshot.GetByID(ctx, f.account) if xerr != nil { return xerr } @@ -29,7 +45,7 @@ func TestLiveMailTransportSnapshotKeepsConfigurationAndGenerationTogether(t *tes return err } defer tx.Rollback(context.Background()) - if _, err := tx.Exec(t.Context(), `UPDATE email_accounts SET transport_generation=$2,name='Fresh' WHERE id=$1`, f.account, next); err != nil { + if _, err := tx.Exec(ctx, `UPDATE email_accounts SET transport_generation=$2,name='Fresh',sync_skip_folders=ARRAY['Later'] WHERE id=$1`, f.account, next); err != nil { return err } if _, err := tx.Exec(t.Context(), `UPDATE email_accounts_oauth SET access_token='fresh-access',refresh_token='fresh-refresh' WHERE email_account_id=$1`, f.account); err != nil { @@ -38,27 +54,35 @@ func TestLiveMailTransportSnapshotKeepsConfigurationAndGenerationTogether(t *tes if err := tx.Commit(t.Context()); err != nil { return err } - credentials, xerr := snapshot.GetOAuthCredentials(t.Context(), f.account) + credentials, xerr := snapshot.GetOAuthCredentials(ctx, f.account) if xerr != nil { return xerr } - if account.TransportGeneration != f.generation || credentials.AccessToken != "old-access" || credentials.RefreshToken != "old-refresh" { + folders, xerr := snapshot.GetSyncSkipFolders(ctx, f.account) + if xerr != nil { + return xerr + } + if account.TransportGeneration != f.generation || credentials.AccessToken != "old-access" || credentials.RefreshToken != "old-refresh" || len(folders) != 1 || folders[0] != "INBOX" { t.Fatal("configuration crossed a generation boundary") } return nil }); err != nil { t.Fatal(err) } - if err := repo.(EmailTransportSnapshot).WithTransportSnapshot(t.Context(), func(snapshot EmailRepository) error { - account, xerr := snapshot.GetByID(t.Context(), f.account) + if err := repo.(EmailTransportSnapshot).WithTransportSnapshot(ctx, func(snapshot EmailRepository) error { + account, xerr := snapshot.GetByID(ctx, f.account) if xerr != nil { return xerr } - credentials, xerr := snapshot.GetOAuthCredentials(t.Context(), f.account) + credentials, xerr := snapshot.GetOAuthCredentials(ctx, f.account) if xerr != nil { return xerr } - if account.TransportGeneration != next || account.Name != "Fresh" || credentials.AccessToken != "fresh-access" { + folders, xerr := snapshot.GetSyncSkipFolders(ctx, f.account) + if xerr != nil { + return xerr + } + if account.TransportGeneration != next || account.Name != "Fresh" || credentials.AccessToken != "fresh-access" || len(folders) != 1 || folders[0] != "Later" { t.Fatal("fresh snapshot did not load fresh configuration") } return nil diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index 1841f1db9..4a7427a72 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -2834,7 +2834,7 @@ func (r *emailRepository) UpdateTrackDirectMail(ctx context.Context, orgID, emai func (r *emailRepository) GetSyncSkipFolders(ctx context.Context, emailAccountID uuid.UUID) ([]string, *errx.Error) { query := `SELECT sync_skip_folders FROM email_accounts WHERE id = $1` var folders []string - if err := r.DB.QueryRow(ctx, query, emailAccountID).Scan(&folders); err != nil { + if err := r.transportQueryRow(ctx, query, emailAccountID).Scan(&folders); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, errx.ErrNotFound }