diff --git a/internal/app/consumer/event_email_error.go b/internal/app/consumer/event_email_error.go index f62a8918..b99a6c0c 100644 --- a/internal/app/consumer/event_email_error.go +++ b/internal/app/consumer/event_email_error.go @@ -51,7 +51,7 @@ func (s *JobsService) HandleEmailAuthError(ctx context.Context, event models.Ema TaskID: taskID, } - if _, xerr := s.EmailAccountErrorRepository.Create(ctx, errorRecord); xerr != nil { + if _, xerr := s.EmailAccountErrorRepository.CreateOnce(ctx, errorRecord); xerr != nil { log.Error().Str("error", xerr.Message).Msg("Failed to store email auth error") } } @@ -116,7 +116,7 @@ func (s *JobsService) HandleEmailDisabled(ctx context.Context, event models.Emai TaskID: taskID, } - if _, xerr := s.EmailAccountErrorRepository.Create(ctx, errorRecord); xerr != nil { + if _, xerr := s.EmailAccountErrorRepository.CreateOnce(ctx, errorRecord); xerr != nil { log.Error().Str("error", xerr.Message).Msg("Failed to store email disabled error") } } @@ -181,7 +181,7 @@ func (s *JobsService) HandleEmailRateLimited(ctx context.Context, event models.E TaskID: taskID, } - if _, xerr := s.EmailAccountErrorRepository.Create(ctx, errorRecord); xerr != nil { + if _, xerr := s.EmailAccountErrorRepository.CreateOnce(ctx, errorRecord); xerr != nil { log.Error().Str("error", xerr.Message).Msg("Failed to store rate limit error") } } diff --git a/internal/app/consumer/event_email_error_test.go b/internal/app/consumer/event_email_error_test.go index b1f27de3..bb6d20a2 100644 --- a/internal/app/consumer/event_email_error_test.go +++ b/internal/app/consumer/event_email_error_test.go @@ -13,16 +13,10 @@ import ( // stubCreateRepo counts which of the two write paths the handler took. type stubCreateRepo struct { repository.EmailAccountErrorRepository - created int createdOnce int suppress bool } -func (s *stubCreateRepo) Create(context.Context, *repository.CreateEmailAccountError) (*repository.EmailAccountError, *errx.Error) { - s.created++ - return &repository.EmailAccountError{}, nil -} - func (s *stubCreateRepo) CreateOnce(_ context.Context, in *repository.CreateEmailAccountError) (*repository.EmailAccountError, *errx.Error) { s.createdOnce++ if s.suppress { @@ -53,9 +47,6 @@ func TestServerErrorRecordsOnePerUnresolvedCode(t *testing.T) { } } - if repo.created != 0 { - t.Errorf("took the unconditional Create path %d times; a repeating server error must not stack rows", repo.created) - } if repo.createdOnce != 3 { t.Errorf("CreateOnce called %d times, want one per relayed failure", repo.createdOnce) } diff --git a/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.down.sql b/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.down.sql new file mode 100644 index 00000000..ed3dfbb5 --- /dev/null +++ b/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.down.sql @@ -0,0 +1,17 @@ +-- Reverses 000145. The rows the up migration collapsed are recoverable +-- because it stamped them with a resolved_by nothing else writes. + +BEGIN; + +CREATE INDEX IF NOT EXISTS idx_email_account_errors_code + ON email_account_errors (email_account_id, error_code) + WHERE resolved_at IS NULL; + +DROP INDEX IF EXISTS idx_email_account_errors_one_unresolved_per_code; + +UPDATE email_account_errors +SET resolved_at = NULL, + resolved_by = NULL +WHERE resolved_by = 'superseded'; + +COMMIT; diff --git a/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.up.sql b/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.up.sql new file mode 100644 index 00000000..3750b714 --- /dev/null +++ b/internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.up.sql @@ -0,0 +1,40 @@ +-- One unresolved error row per mailbox per code. +-- +-- The sync loop retries a refused mail server about once a minute and relays +-- what it got every time, and the consumer recorded each one, so a mailbox +-- whose server kept saying no collected an identical row a minute for as long +-- as it lasted (issue #405). The insert is conditional now, but a conditional +-- insert is not atomic on its own: two workers relaying the same failure at +-- the same moment both see no row and both write one. +-- +-- The unique index is what actually holds. It is partial on the unresolved +-- rows, so the history of resolved errors is untouched and a problem that +-- comes back after it was fixed is still recorded as news. + +BEGIN; + +-- Existing duplicates first, or the index cannot be built. The oldest of each +-- group is kept because it says when the problem started; the rest are marked +-- resolved rather than deleted, so the record of how long it went on survives. +UPDATE email_account_errors AS e +SET resolved_at = NOW(), + resolved_by = 'superseded' +WHERE e.resolved_at IS NULL + AND EXISTS ( + SELECT 1 + FROM email_account_errors AS keep + WHERE keep.email_account_id = e.email_account_id + AND keep.error_code = e.error_code + AND keep.resolved_at IS NULL + AND (keep.created_at, keep.id) < (e.created_at, e.id) + ); + +CREATE UNIQUE INDEX idx_email_account_errors_one_unresolved_per_code + ON email_account_errors (email_account_id, error_code) + WHERE resolved_at IS NULL; + +-- Same columns, same predicate, no longer unique: the index above serves every +-- lookup this one did, and a redundant index is paid for on every write. +DROP INDEX IF EXISTS idx_email_account_errors_code; + +COMMIT; diff --git a/internal/repository/email_error_dedupe_live_test.go b/internal/repository/email_error_dedupe_live_test.go index da521679..de88d20a 100644 --- a/internal/repository/email_error_dedupe_live_test.go +++ b/internal/repository/email_error_dedupe_live_test.go @@ -2,6 +2,7 @@ package repository import ( "context" + "sync" "testing" "github.com/google/uuid" @@ -121,3 +122,94 @@ func TestLiveEmailErrorDedupeKeepsOneUnresolvedRowPerCode(t *testing.T) { t.Error("an error recurring after it was resolved must be recorded again") } } + +// CodeRabbit's catch on #406: a conditional insert is not atomic on its own. +// Two workers relaying the same failure in the same instant both find no row +// and both write one. Migration 000145 makes that impossible and ON CONFLICT +// turns the loser into the ordinary "already on screen" answer instead of a +// logged storage failure. +func TestLiveEmailErrorDedupeSurvivesConcurrentRelays(t *testing.T) { + handle, pool := liveContactDB(t) + repo := NewEmailAccountErrorRepository(handle) + ctx := context.Background() + + account, user, org := uuid.New(), uuid.New(), uuid.New() + tag := "i405c-" + org.String()[:8] + exec := func(sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture: %v", err) + } + } + exec(`INSERT INTO users (id, first_name, last_name, email, password_hash) + VALUES ($1, 'Ilse', 'Live', $2, 'x')`, user, tag+"@test.local") + exec(`INSERT INTO organizations (id, name, slug, owner_user_id) + VALUES ($1, 'Issue 405', $2, $3)`, org, tag, user) + exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, signature_html, provider) + VALUES ($1, $2, $3, $4, 'Ilse', '', '', 'smtp_imap')`, + account, user, org, tag+"-mb@test.local") + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM email_account_errors WHERE email_account_id = $1`, account}, + {`DELETE FROM email_accounts WHERE organization_id = $1`, org}, + {`DELETE FROM organizations WHERE id = $1`, org}, + {`DELETE FROM users WHERE id = $1`, user}, + } { + if _, err := pool.Exec(c, step.sql, step.arg); err != nil { + t.Errorf("cleanup: %v", err) + } + } + }) + + const racers = 8 + var ( + wg sync.WaitGroup + mu sync.Mutex + written int + ) + start := make(chan struct{}) + for i := 0; i < racers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + row, xerr := repo.CreateOnce(ctx, &CreateEmailAccountError{ + EmailAccountID: account, + UserID: user, + ErrorCode: "IMAP_UNKNOWN", + Severity: "WARNING", + ResolveMethod: "RETRY", + Title: "Email Error", + Message: "Something went wrong: NO System Error", + }) + mu.Lock() + defer mu.Unlock() + if xerr != nil { + t.Errorf("a losing racer must not surface a storage failure: %v", xerr.Message) + return + } + if row != nil { + written++ + } + }() + } + close(start) + wg.Wait() + + if written != 1 { + t.Errorf("%d of %d racers believed they wrote the row, want exactly 1", written, racers) + } + var n int + if err := pool.QueryRow(ctx, + `SELECT count(*) FROM email_account_errors + WHERE email_account_id = $1 AND resolved_at IS NULL`, account).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n != 1 { + t.Errorf("unresolved rows = %d after %d concurrent relays, want 1", n, racers) + } +} diff --git a/internal/repository/pg_email_error.go b/internal/repository/pg_email_error.go index d9172cd7..31b15542 100644 --- a/internal/repository/pg_email_error.go +++ b/internal/repository/pg_email_error.go @@ -45,10 +45,11 @@ type CreateEmailAccountError struct { // EmailAccountErrorRepository defines operations for email account errors type EmailAccountErrorRepository interface { - Create(ctx context.Context, err *CreateEmailAccountError) (*EmailAccountError, *errx.Error) - // CreateOnce is Create for an error that repeats until someone fixes it: - // it records nothing, and returns a nil record, while the account already - // has an unresolved error with the same code. + // CreateOnce records an error unless the account already has an unresolved + // one with the same code, in which case it writes nothing and returns a + // nil record. It is the only write path: migration 000145 makes a second + // unresolved row for one code impossible, so an unconditional insert would + // only turn a repeat into a constraint violation. CreateOnce(ctx context.Context, err *CreateEmailAccountError) (*EmailAccountError, *errx.Error) GetByAccountID(ctx context.Context, accountID uuid.UUID, unresolvedOnly bool) ([]EmailAccountError, *errx.Error) GetByUserID(ctx context.Context, userID uuid.UUID, limit int) ([]EmailAccountError, *errx.Error) @@ -76,53 +77,20 @@ func NewEmailAccountErrorRepository(database *db.DB) EmailAccountErrorRepository } // Create stores a new email account error -func (r *emailAccountErrorRepository) Create(ctx context.Context, data *CreateEmailAccountError) (*EmailAccountError, *errx.Error) { - query := ` - INSERT INTO email_account_errors ( - email_account_id, user_id, error_code, severity, resolve_method, - title, message, user_message, action_required, task_id - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) - RETURNING id, email_account_id, user_id, error_code, severity, resolve_method, - title, message, user_message, action_required, task_id, - resolved_at, resolved_by, created_at - ` - - params := []any{ - data.EmailAccountID, - data.UserID, - data.ErrorCode, - data.Severity, - data.ResolveMethod, - data.Title, - data.Message, - data.UserMessage, - data.ActionRequired, - data.TaskID, - } - - var e EmailAccountError - err := r.DB.QueryRow(ctx, query, params...).Scan( - &e.ID, &e.EmailAccountID, &e.UserID, &e.ErrorCode, &e.Severity, &e.ResolveMethod, - &e.Title, &e.Message, &e.UserMessage, &e.ActionRequired, &e.TaskID, - &e.ResolvedAt, &e.ResolvedBy, &e.CreatedAt, - ) - if err != nil { - db.CaptureError(err, query, params, "queryrow") - return nil, errx.InternalError() - } - - return &e, nil -} - // CreateOnce records the error unless the account already has an unresolved // one carrying the same code, and returns (nil, nil) when it declined. // // A mail server that refuses the same command every pass is not a new problem // every pass. The sync loop retries about once a minute and reports what it // got, so one IMAP_UNKNOWN that nobody can fix wrote 1440 identical rows a day -// into the mailbox's error list (issue #405). The insert is conditional in SQL -// rather than a read followed by a write, because several workers can relay -// the same failure at once. +// into the mailbox's error list (issue #405). +// +// Both guards are load-bearing. WHERE NOT EXISTS answers the ordinary repeat +// without touching the index, and ON CONFLICT answers the race it cannot see: +// two workers relaying the same failure in the same instant both find no row. +// The unique index behind it is partial on the unresolved rows (migration +// 000145), so resolved history is untouched and a problem that returns after +// it was fixed is recorded again. func (r *emailAccountErrorRepository) CreateOnce(ctx context.Context, data *CreateEmailAccountError) (*EmailAccountError, *errx.Error) { query := ` INSERT INTO email_account_errors ( @@ -138,6 +106,7 @@ func (r *emailAccountErrorRepository) CreateOnce(ctx context.Context, data *Crea SELECT 1 FROM email_account_errors WHERE email_account_id = $1 AND error_code = $3 AND resolved_at IS NULL ) + ON CONFLICT DO NOTHING RETURNING id, email_account_id, user_id, error_code, severity, resolve_method, title, message, user_message, action_required, task_id, resolved_at, resolved_by, created_at