starred/warmbly
1
0
Fork 0
mirror of https://github.com/warmbly/warmbly.git synced 2026-09-12 16:04:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: make the one-unresolved-error-per-code rule a partial unique index with a conflict path, collapsing the duplicate rows a repeating mail error already left behind

Browse Source
This commit is contained in:
Matthew Meszaros
2026-09-09 09:46:39 -07:00
parent 817233cc09
commit bb2b2b1743
6 changed files with 166 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/consumer/event_email_error.go
+3 -3
View File
@@ -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")
}
}
internal/app/consumer/event_email_error_test.go
-9
View File
@@ -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)
}
internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.down.sql
+17
View File
@@ -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;
internal/infrastructure/db/migrations/000145_one_unresolved_error_per_code.up.sql
+40
View File
@@ -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;
internal/repository/email_error_dedupe_live_test.go
+92
View File
@@ -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)
}
}
internal/repository/pg_email_error.go
+14 -45
View File
@@ -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