feat: fence single-connection mailbox loads and preserve typed sync truth, pre-upgrade Avro fixtures, and mandatory transport regression execution

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

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
 		}

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 {

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 = &copy
+	} 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 {

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

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

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

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

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)

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

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

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
 		}