diff --git a/services/protocols/internal/application/usecase/sent_copy_test.go b/services/protocols/internal/application/usecase/sent_copy_test.go index 019783d..52c5146 100644 --- a/services/protocols/internal/application/usecase/sent_copy_test.go +++ b/services/protocols/internal/application/usecase/sent_copy_test.go @@ -137,3 +137,51 @@ func TestSubmit_WorksWithoutASentCopyRepository(t *testing.T) { t.Fatalf("Submit without a Sent repository: %v", err) } } + +// One rule for every path: a message sent from the webmail and the same message +// sent by an SMTP client leave the same thing behind in Sent. They used to differ +// completely - the webmail filed a copy, SMTP filed nothing - because the rule +// lived in the webmail alone. +func TestSentCopyIsTheSameWhicheverPathSent(t *testing.T) { + mailboxID := uuid.New() + messages := &recordingMessageRepository{} + submission := NewProcessOutboundEmailUseCase(nil, &countingOutboundRepo{}, &capturingBlobStorage{}, nil) + submission.SetSentCopies(messages) + + account := &port.MailboxAuth{ID: mailboxID, EmailAddress: "me@example.test", DomainName: "example.test", MaxRecipientsPerHour: 100} + web := &WebmailUseCase{ + repo: &stubWebmailRepo{mailboxID: mailboxID.String()}, + auth: &stubAuthRepo{account: account}, + submission: submission, + localHost: "mail.example.test", + } + + if err := web.Send(context.Background(), ComposeInput{From: "me@example.test", To: []string{"you@remote.test"}, Subject: "Aviso", Text: "hi"}); err != nil { + t.Fatalf("webmail send: %v", err) + } + + smtp := ProcessOutboundEmailInput{ + MailboxID: mailboxID, SenderAddr: "me@example.test", DomainName: "example.test", + Recipients: []string{"you@remote.test"}, FileSentCopy: true, + Body: bytes.NewBufferString("From: me@example.test\r\nTo: you@remote.test\r\nSubject: Aviso\r\n\r\nhi"), + } + if err := submission.Submit(context.Background(), smtp); err != nil { + t.Fatalf("smtp submit: %v", err) + } + + if len(messages.persisted) != 2 { + t.Fatalf("want one copy per path (2), got %d: the webmail must not file its own copy as well", len(messages.persisted)) + } + fromWeb, fromSMTP := messages.persisted[0], messages.persisted[1] + for _, got := range []port.PersistInboundMessageInput{fromWeb, fromSMTP} { + if got.TargetFolderName != "Sent" || !got.AlreadySeen || got.MailboxID != mailboxID { + t.Errorf("a Sent copy is not what the rule says: %+v", got) + } + if got.SenderAddress != "me@example.test" || got.RecipientAddress != "me@example.test" { + t.Errorf("a Sent copy belongs to its sender: %+v", got) + } + if got.Subject != "Aviso" { + t.Errorf("subject = %q, want Aviso", got.Subject) + } + } +} diff --git a/services/protocols/internal/application/usecase/webmail_compose_test.go b/services/protocols/internal/application/usecase/webmail_compose_test.go index 161abf0..0b7dffd 100644 --- a/services/protocols/internal/application/usecase/webmail_compose_test.go +++ b/services/protocols/internal/application/usecase/webmail_compose_test.go @@ -197,6 +197,8 @@ func newSendableUseCaseWith( } submission := NewProcessOutboundEmailUseCase( nil, &countingOutboundRepo{}, &capturingBlobStorage{}, &recordingSigner{}) + // The Sent copy is filed by the submission, as it is for SMTP clients + submission.SetSentCopies(messages) return (&WebmailUseCase{ repo: repo, auth: &stubAuthRepo{account: account}, diff --git a/services/protocols/internal/application/usecase/webmail_mail.go b/services/protocols/internal/application/usecase/webmail_mail.go index 361c9f2..f67c0af 100644 --- a/services/protocols/internal/application/usecase/webmail_mail.go +++ b/services/protocols/internal/application/usecase/webmail_mail.go @@ -32,10 +32,9 @@ type WebmailUseCase struct { submission *ProcessOutboundEmailUseCase auth port.AuthRepository localHost string - // blobStore and messages file the sender's own copy into Sent, and hold - // drafts. Both are optional: when either is nil the copy is skipped rather - // than the send failing, because a message that has already been queued - // for delivery must not be reported as an error. + // blobStore and messages hold drafts. Both are optional: when either is nil + // saving a draft reports ErrDraftsUnavailable. (The copy in Sent is filed by + // the submission use case, not here.) blobStore port.BlobStorage messages port.InboundMessageRepository } @@ -53,7 +52,7 @@ func NewWebmailUseCase( return &WebmailUseCase{repo: repo, blobs: blobs, submission: submission, auth: auth, localHost: localHost} } -// WithLocalFiling enables the Sent copy and draft storage. +// WithLocalFiling enables draft storage. func (uc *WebmailUseCase) WithLocalFiling(store port.BlobStorage, messages port.InboundMessageRepository) *WebmailUseCase { uc.blobStore = store uc.messages = messages @@ -188,6 +187,10 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { payload := buildMimeMessage(input, recipients, uc.localHost) + // The copy in Sent is filed by Submit, the same way it is for a client that + // sends over SMTP: one rule, one place. It used to be filed here, after the + // submission, and not at all for SMTP, so an account that sent by SMTP had an + // empty Sent folder. if err := uc.submission.Submit(ctx, ProcessOutboundEmailInput{ MailboxID: account.ID, SenderAddr: account.EmailAddress, @@ -195,19 +198,11 @@ func (uc *WebmailUseCase) Send(ctx context.Context, input ComposeInput) error { Recipients: recipients, Body: bytes.NewReader(payload), MaxRecipientsPerHour: account.MaxRecipientsPerHour, + FileSentCopy: true, }); err != nil { return err } - // Only after the message is durably queued. Filing it first would leave a - // copy in Sent for a message that was then rejected by the send limit. - // - // A failure here is logged and swallowed: the mail is already on its way, - // and telling the composer the send failed would invite a duplicate. - if err := uc.fileLocalCopy(ctx, account.ID, account.EmailAddress, "Sent", payload); err != nil { - log.Printf("webmail: could not file the Sent copy for %s: %v", account.EmailAddress, err) - } - // The draft this message grew out of has been superseded by the real // thing. Swallowed like the Sent copy above and for the same reason: the // mail is already queued, and failing the request now would only invite a @@ -288,36 +283,6 @@ func (uc *WebmailUseCase) Delete(ctx context.Context, address, folder string, ui } } -// fileLocalCopy stores a message the user themselves composed into one of -// their own folders. It reuses the inbound persistence path so the copy gets a -// real IMAP UID and shows up over IMAP as well as in the webmail. -func (uc *WebmailUseCase) fileLocalCopy( - ctx context.Context, mailboxID uuid.UUID, address, folder string, payload []byte, -) error { - if uc.blobStore == nil || uc.messages == nil { - return nil - } - blob, err := uc.blobStore.Store(ctx, bytes.NewReader(payload)) - if err != nil { - return err - } - headers := ExtractMessageHeaders(payload) - _, err = uc.messages.PersistAll(ctx, []port.PersistInboundMessageInput{{ - MailboxID: mailboxID, - Blob: blob, - SenderAddress: address, - RecipientAddress: address, - TargetFolderName: folder, - Subject: headers.Subject, - Snippet: headers.Snippet, - FromDisplayName: headers.FromDisplayName, - MessageIDHeader: headers.MessageID, - HasAttachments: headers.HasAttachments, - AlreadySeen: true, - }}) - return err -} - // SaveDraft files the message being composed into the Drafts folder. // // Nothing did this before: the composer displayed "draft saved automatically" diff --git a/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go new file mode 100644 index 0000000..3153434 --- /dev/null +++ b/services/protocols/internal/e2e/one_mailbox_everywhere_e2e_test.go @@ -0,0 +1,385 @@ +package e2e + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alexedwards/argon2id" + "github.com/emersion/go-imap/v2" + "github.com/emersion/go-imap/v2/imapclient" + "github.com/emersion/go-imap/v2/imapserver" + gosmtp "github.com/emersion/go-smtp" + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/google/uuid" + + "lambdamail/protocols/internal/application/port" + appusecase "lambdamail/protocols/internal/application/usecase" + "lambdamail/protocols/internal/infrastructure/diskstorage" + "lambdamail/protocols/internal/infrastructure/dkim" + "lambdamail/protocols/internal/infrastructure/postgres" + httppresentation "lambdamail/protocols/internal/presentation/http" + imappresentation "lambdamail/protocols/internal/presentation/imap" + smtppresentation "lambdamail/protocols/internal/presentation/smtp" +) + +// TestOneMailboxEverywhere is the contract of the whole server: SMTP, IMAP and +// the webmail are three ways into the same mailbox, so whatever one does, the +// other two must see, with the same numbers. +// +// The webmail used to carry its own copy of the rules and its own queries, and +// the two stacks drifted apart: mail sent over SMTP never reached Sent, reading +// over IMAP never moved the unread badge, deleting in the webmail never gave the +// quota back. This test drives all three on one mailbox, mixing them, and checks +// after every step that each view agrees with the others and with the data. +func TestOneMailboxEverywhere(t *testing.T) { + if testing.Short() { + t.Skip("e2e test skipped in -short mode") + } + ctx := context.Background() + + pg := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + Port(15451). + RuntimePath(t.TempDir()). + StartTimeout(120 * time.Second)) + if err := pg.Start(); err != nil { + t.Fatalf("start embedded postgres: %v", err) + } + defer pg.Stop() + + pool, err := postgres.NewPool(ctx, "postgres://postgres:postgres@localhost:15451/postgres?sslmode=disable") + if err != nil { + t.Fatalf("connect pool: %v", err) + } + defer pool.Close() + applyMigrations(t, ctx, pool) + + // ------------------------------------------------------------- fixtures + const ( + address = "plataforma@everywhere.test" + password = "a long enough password" + ) + hash, err := argon2id.CreateHash(password, argon2id.DefaultParams) + if err != nil { + t.Fatalf("hash: %v", err) + } + domainID, mailboxID := uuid.New(), uuid.New() + pool.Exec(ctx, `INSERT INTO domains (id, name, punycode_name) VALUES ($1, 'everywhere.test', 'everywhere.test')`, domainID) + pool.Exec(ctx, `INSERT INTO mailboxes (id, domain_id, local_part, email_address, password_hash) VALUES ($1, $2, 'plataforma', $3, $4)`, + mailboxID, domainID, address, hash) + // The standard folders come from the one definition in the database + if _, err := pool.Exec(ctx, `SELECT provision_default_folders($1)`, mailboxID); err != nil { + t.Fatalf("provision folders: %v", err) + } + + blobs := diskstorage.NewLocalDiskBlobStorage(pool, t.TempDir()) + blobReader := diskstorage.NewLocalDiskBlobReader(pool) + authRepo := postgres.NewAuthRepository(pool) + messages := postgres.NewInboundMessageRepository(pool) + + // Three messages arrive from outside, as the SMTP side would file them + inbound := appusecase.NewProcessInboundEmailUseCase(postgres.NewMailboxRepository(pool), blobs, messages) + target, err := postgres.NewMailboxRepository(pool).FindActiveByAddress(ctx, address) + if err != nil || target == nil { + t.Fatalf("resolve mailbox: %v", err) + } + for i := 1; i <= 3; i++ { + raw := fmt.Sprintf("From: alice@remote.test\r\nTo: %s\r\nSubject: inbound %d\r\nMessage-ID: \r\n\r\nbody %d\r\n", address, i, i, i) + if err := inbound.Handle(ctx, appusecase.ProcessInboundEmailInput{ + Sender: "alice@remote.test", Recipients: []port.MailboxRecord{*target}, RecipientAddresses: []string{address}, Body: strings.NewReader(raw), + }); err != nil { + t.Fatalf("deliver %d: %v", i, err) + } + } + + // -------------------------------------------- the three ways in, wired as main.go + submissionUC := appusecase.NewProcessOutboundEmailUseCase(authRepo, postgres.NewOutboundRepository(pool), blobs, dkim.NewSigner(nil)) + submissionUC.SetSentCopies(messages) + + smtpServer := gosmtp.NewServer(smtppresentation.NewSubmissionBackend(submissionUC)) + smtpServer.Domain = "mail.everywhere.test" + smtpServer.TLSConfig = &tls.Config{Certificates: []tls.Certificate{selfSignedTLSCert(t)}} + smtpLn, _ := net.Listen("tcp", "127.0.0.1:0") + defer smtpLn.Close() + go smtpServer.Serve(smtpLn) + + imapUC := appusecase.NewImapSessionUseCase( + authRepo, postgres.NewImapFolderRepository(pool), postgres.NewMessageQueryRepository(pool), + postgres.NewFlagRepository(pool), blobReader, postgres.NewExpungeRepository(pool), postgres.NewCopyRepository(pool)) + imapServer := imapserver.New(&imapserver.Options{ + NewSession: func(c *imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) { + return imappresentation.NewSession(c, imapUC) + }, + Caps: imap.CapSet{imap.CapIMAP4rev1: {}, imap.CapMove: {}, imap.CapUIDPlus: {}}, + InsecureAuth: true, + }) + imapLn, _ := net.Listen("tcp", "127.0.0.1:0") + go imapServer.Serve(imapLn) + defer imapServer.Close() + + webmailUC := appusecase.NewWebmailUseCase(postgres.NewWebmailRepository(pool), blobReader, submissionUC, authRepo, "mail.everywhere.test"). + WithLocalFiling(blobs, messages) + router := httppresentation.NewRouter(nil, func() error { return nil }) + router.SetMailAPI(webmailUC, webmailSecret) + web := httptest.NewServer(router) + defer web.Close() + token := mintSession(t, address, "user", time.Hour) + + // --------------------------------------------------------- small drivers + webCall := func(method, path string, body any) []byte { + t.Helper() + var reader *bytes.Reader + if body != nil { + encoded, _ := json.Marshal(body) + reader = bytes.NewReader(encoded) + } else { + reader = bytes.NewReader(nil) + } + req, _ := http.NewRequest(method, web.URL+path, reader) + req.Header.Set("Cookie", "lm_user_session="+token) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("webmail %s %s: %v", method, path, err) + } + defer resp.Body.Close() + buf := new(bytes.Buffer) + buf.ReadFrom(resp.Body) + if resp.StatusCode >= 300 { + t.Fatalf("webmail %s %s = %d: %s", method, path, resp.StatusCode, buf.String()) + } + return buf.Bytes() + } + + type webFolder struct { + Name string `json:"name"` + UnreadCount int `json:"unread_count"` + TotalCount int `json:"total_count"` + } + webFolders := func() map[string]webFolder { + t.Helper() + var payload struct { + Folders []webFolder `json:"folders"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/folders", nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Folders == nil { + var plain []webFolder + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode folders: %v / %v: %s", err, err2, raw) + } + payload.Folders = plain + } + out := map[string]webFolder{} + for _, f := range payload.Folders { + out[f.Name] = f + } + return out + } + type webMessage struct { + UID uint32 `json:"uid"` + Subject string `json:"subject"` + Seen bool `json:"seen"` + } + webMessages := func(folder string) []webMessage { + t.Helper() + var payload struct { + Messages []webMessage `json:"messages"` + } + raw := webCall(http.MethodGet, "/api/v1/mail/messages?folder="+folder, nil) + if err := json.Unmarshal(raw, &payload); err != nil || payload.Messages == nil { + var plain []webMessage + if err2 := json.Unmarshal(raw, &plain); err2 != nil { + t.Fatalf("decode messages: %v / %v: %s", err, err2, raw) + } + payload.Messages = plain + } + return payload.Messages + } + + imapClient, err := imapclient.DialInsecure(imapLn.Addr().String(), nil) + if err != nil { + t.Fatalf("dial imap: %v", err) + } + defer imapClient.Close() + if err := imapClient.Login(address, password).Wait(); err != nil { + t.Fatalf("imap login: %v", err) + } + imapExists := func(folder string) int { + t.Helper() + data, err := imapClient.Select(folder, nil).Wait() + if err != nil { + t.Fatalf("imap select %s: %v", folder, err) + } + return int(data.NumMessages) + } + imapFlags := func(folder string) map[uint32][]imap.Flag { + t.Helper() + imapExists(folder) + all := imap.UIDSet{} + all.AddRange(1, 0) // 1:* + cmd := imapClient.Fetch(all, &imap.FetchOptions{UID: true, Flags: true}) + out := map[uint32][]imap.Flag{} + msgs, err := cmd.Collect() + if err != nil { + t.Fatalf("imap fetch flags in %s: %v", folder, err) + } + for _, m := range msgs { + out[uint32(m.UID)] = m.Flags + } + return out + } + + // agree compares what the webmail and IMAP each say about every folder, and + // both with the rows: the one thing that must never differ + agree := func(after string) { + t.Helper() + assertEverywhere := webFolders() + for _, name := range []string{"INBOX", "Sent", "Trash", "Drafts"} { + if got, want := imapExists(name), assertEverywhere[name].TotalCount; got != want { + t.Errorf("after %s: %s holds %d messages over IMAP and %d in the webmail", after, name, got, want) + } + } + var badFolders, badBytes int + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM ( + SELECT f.id FROM folders f + LEFT JOIN email_messages e ON e.folder_id = f.id AND e.expunged_at IS NULL + WHERE f.mailbox_id = $1 + GROUP BY f.id, f.total_count, f.unread_count + HAVING f.total_count <> COUNT(e.id) + OR f.unread_count <> COUNT(e.id) FILTER (WHERE NOT EXISTS ( + SELECT 1 FROM message_flags mf WHERE mf.message_id = e.id AND mf.received_at = e.received_at AND mf.flag = '\Seen')) + ) x`, mailboxID).Scan(&badFolders) + pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM mailboxes m + WHERE m.id = $1 AND m.used_bytes <> COALESCE((SELECT SUM(size_bytes) FROM email_messages WHERE mailbox_id = m.id AND expunged_at IS NULL), 0)`, + mailboxID).Scan(&badBytes) + if badFolders != 0 { + t.Errorf("after %s: %d folder(s) have cached counters that differ from their messages", after, badFolders) + } + if badBytes != 0 { + t.Errorf("after %s: the quota differs from the bytes actually stored", after) + } + } + + agree("delivery") + if f := webFolders(); f["INBOX"].UnreadCount != 3 || f["INBOX"].TotalCount != 3 { + t.Fatalf("INBOX should start with 3 unread of 3, the webmail says %+v", f["INBOX"]) + } + + // ----- 1. SMTP sends a message; Sent must show it in the webmail and over IMAP + smtpClient, err := gosmtp.DialStartTLS(smtpLn.Addr().String(), &tls.Config{InsecureSkipVerify: true}) + if err != nil { + t.Fatalf("dial smtp: %v", err) + } + if err := smtpClient.Auth(plainAuth(address, password)); err != nil { + t.Fatalf("smtp auth: %v", err) + } + smtpClient.Mail(address, nil) + smtpClient.Rcpt("bob@remote.test", nil) + w, _ := smtpClient.Data() + w.Write([]byte("From: " + address + "\r\nTo: bob@remote.test\r\nSubject: sent over smtp\r\nMessage-ID: \r\n\r\nhello\r\n")) + if err := w.Close(); err != nil { + t.Fatalf("smtp data: %v", err) + } + smtpClient.Quit() + + agree("sending over SMTP") + if sent := webMessages("sent"); len(sent) != 1 || sent[0].Subject != "sent over smtp" { + t.Fatalf("the webmail's Sent should hold the SMTP message, got %+v", sent) + } + if n := imapExists("Sent"); n != 1 { + t.Fatalf("IMAP's Sent holds %d messages after an SMTP send, want 1", n) + } + if f := webFolders(); f["Sent"].UnreadCount != 0 { + t.Errorf("the sender's own copy shows as unread: %+v", f["Sent"]) + } + + // ----- 2. the webmail sends one too: it must land in the same place, once + webCall(http.MethodPost, "/api/v1/mail/send", map[string]any{"to": []string{"carol@remote.test"}, "subject": "sent from webmail", "text": "hi"}) + agree("sending from the webmail") + if n := imapExists("Sent"); n != 2 { + t.Fatalf("Sent holds %d messages after one SMTP send and one webmail send, want exactly 2", n) + } + + // ----- 3. read in the webmail; IMAP must see \Seen and the badge must fall + inbox := webMessages("inbox") + if len(inbox) != 3 { + t.Fatalf("webmail lists %d INBOX messages, want 3", len(inbox)) + } + uidA, uidB, uidC := inbox[2].UID, inbox[1].UID, inbox[0].UID // oldest first by UID + webCall(http.MethodPost, "/api/v1/mail/seen", map[string]any{"folder": "inbox", "uid": uidA, "seen": true}) + agree("reading a message in the webmail") + if flags := imapFlags("INBOX")[uidA]; !hasFlag(flags, imap.FlagSeen) { + t.Errorf("IMAP does not see the message the webmail marked as read: %v", flags) + } + if f := webFolders(); f["INBOX"].UnreadCount != 2 { + t.Errorf("INBOX unread = %d after one read, want 2", f["INBOX"].UnreadCount) + } + + // ----- 4. read over IMAP; the webmail's badge must follow (it never did) + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidB)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagSeen}}, nil).Close(); err != nil { + t.Fatalf("imap store: %v", err) + } + agree("reading a message over IMAP") + if f := webFolders(); f["INBOX"].UnreadCount != 1 { + t.Errorf("INBOX unread = %d after a read over IMAP, want 1: the badge ignored it", f["INBOX"].UnreadCount) + } + for _, m := range webMessages("inbox") { + if m.UID == uidB && !m.Seen { + t.Error("the webmail lists the message IMAP marked as read as unread") + } + } + + // ----- 5. delete over IMAP (\Deleted + EXPUNGE); the webmail list shrinks, the quota is returned + imapExists("INBOX") + if err := imapClient.Store(imap.UIDSetNum(imap.UID(uidC)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Silent: true, Flags: []imap.Flag{imap.FlagDeleted}}, nil).Close(); err != nil { + t.Fatalf("imap store deleted: %v", err) + } + if err := imapClient.Expunge().Close(); err != nil { + t.Fatalf("imap expunge: %v", err) + } + agree("deleting over IMAP") + if n := len(webMessages("inbox")); n != 2 { + t.Errorf("the webmail lists %d INBOX messages after an IMAP delete, want 2", n) + } + + // ----- 6. delete in the webmail (to Trash); IMAP sees it in Trash and not in INBOX + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "inbox", "uid": uidA}) + agree("deleting in the webmail") + if n := imapExists("Trash"); n != 1 { + t.Errorf("IMAP's Trash holds %d messages, want 1", n) + } + if n := imapExists("INBOX"); n != 1 { + t.Errorf("IMAP's INBOX holds %d messages, want 1", n) + } + + // ----- 7. empty the Trash in the webmail: the bytes go back to the quota + trash := webMessages("trash") + if len(trash) != 1 { + t.Fatalf("the webmail lists %d Trash messages, want 1", len(trash)) + } + webCall(http.MethodPost, "/api/v1/mail/delete", map[string]any{"folder": "trash", "uid": trash[0].UID}) + agree("emptying the Trash") + if n := imapExists("Trash"); n != 0 { + t.Errorf("IMAP's Trash holds %d messages after emptying it, want 0", n) + } +} + +func hasFlag(flags []imap.Flag, want imap.Flag) bool { + for _, f := range flags { + if f == want { + return true + } + } + return false +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go index fb80f66..54f041c 100644 --- a/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go +++ b/services/protocols/internal/infrastructure/postgres/mail_state_contract_test.go @@ -403,3 +403,132 @@ func TestContract_ReplacingTheFlagSetKeepsTheUnreadCounterTrue(t *testing.T) { } var _ = uuid.Nil + +// IMAP clients, the webmail and the delivery path all touch a mailbox at the same +// moment. The counters are shared rows, so they have to stay true under that. +func TestContract_ConcurrentChangesFromBothPathsKeepTheCountersTrue(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const n = 24 + uids := make([]uint32, 0, n) + for i := 0; i < n; i++ { + uids = append(uids, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("m%d", i), false)) + } + + imap, web := imapPath(), webmailPath() + done := make(chan struct{}, n) + for i, uid := range uids { + go func(i int, uid uint32) { + defer func() { done <- struct{}{} }() + switch i % 3 { + case 0: + imap.setSeen(t, pool, mailboxID, "INBOX", uid, true) + case 1: + web.setSeen(t, pool, mailboxID, "INBOX", uid, true) + default: + if i%2 == 0 { + imap.expunge(t, pool, mailboxID, "INBOX", uid) + } else { + web.expunge(t, pool, mailboxID, "INBOX", uid) + } + } + }(i, uid) + } + for range uids { + <-done + } + + assertConsistent(t, pool, mailboxID, "24 concurrent changes from IMAP and the webmail") +} + +// Two clients deleting the same message at once is the realistic race (the +// webmail and a phone both open on the same mail). It is removed once, and every +// number moves once. +func TestContract_DeletingTheSameMessageConcurrentlyCountsOnce(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + target := deliver(t, pool, mailboxID, "INBOX", "contested", false) + deliver(t, pool, mailboxID, "INBOX", "bystander", false) + inbox := folderID(t, pool, mailboxID, "INBOX") + + ctx := context.Background() + store := NewMailStore(pool) + removed := make(chan int, 12) + for i := 0; i < 12; i++ { + go func() { + n, err := store.Expunge(ctx, inbox, []uint32{target}) + if err != nil { + t.Errorf("expunge: %v", err) + } + removed <- n + }() + } + total := 0 + for i := 0; i < 12; i++ { + total += <-removed + } + + if total != 1 { + t.Errorf("12 concurrent expunges of one message removed it %d times, want exactly once", total) + } + assertConsistent(t, pool, mailboxID, "12 clients deleting the same message") + folders, used, _ := mailState(t, pool, mailboxID) + if folders["INBOX"].RealTotal != 1 || used != 10 { + t.Errorf("INBOX holds %d and the mailbox %d bytes, want 1 message and 10 bytes", folders["INBOX"].RealTotal, used) + } +} + +// Moves in opposite directions, flag changes and deletes, all at once on the same +// two folders. Locking the folders and the message in different orders in +// different operations deadlocks here; a fixed order does not. +func TestContract_OppositeMovesAndFlagChangesNeverDeadlock(t *testing.T) { + pool := testPool(t) + mailboxID := seedMailbox(t, pool) + + const each = 20 + var inboxUIDs, trashUIDs []uint32 + for i := 0; i < each; i++ { + inboxUIDs = append(inboxUIDs, deliver(t, pool, mailboxID, "INBOX", fmt.Sprintf("i%d", i), false)) + trashUIDs = append(trashUIDs, deliver(t, pool, mailboxID, "Trash", fmt.Sprintf("t%d", i), false)) + } + + inbox, trash := folderID(t, pool, mailboxID, "INBOX"), folderID(t, pool, mailboxID, "Trash") + store := NewMailStore(pool) + ctx := context.Background() + + errs := make(chan error, each*4) + var pending int + run := func(f func() error) { + pending++ + go func() { errs <- f() }() + } + for i := 0; i < each; i++ { + i := i + run(func() error { _, err := store.Move(ctx, inbox, inboxUIDs[i], trash); return nil2(err) }) + run(func() error { _, err := store.Move(ctx, trash, trashUIDs[i], inbox); return nil2(err) }) + // a flag change on a message that is being moved elsewhere, and a delete + run(func() error { + _, err := store.SetFlags(ctx, inbox, inboxUIDs[(i+1)%each], port.FlagOpAdd, []string{`\Flagged`}, 0) + return nil2(err) + }) + run(func() error { _, err := store.Expunge(ctx, trash, []uint32{trashUIDs[(i+2)%each]}); return nil2(err) }) + } + for ; pending > 0; pending-- { + if err := <-errs; err != nil { + t.Errorf("a concurrent operation failed (a deadlock, if it says 40P01): %v", err) + } + } + + assertConsistent(t, pool, mailboxID, "opposite moves, flag changes and deletes all at once") +} + +// nil2 lets the contract tolerate "the message was moved or deleted first", a +// legitimate outcome of a race, while still surfacing every other error (above +// all a deadlock, SQLSTATE 40P01). +func nil2(err error) error { + if err == ErrMessageNotFound { + return nil + } + return err +} diff --git a/services/protocols/internal/infrastructure/postgres/mail_store.go b/services/protocols/internal/infrastructure/postgres/mail_store.go index f529dc5..1a0fb30 100644 --- a/services/protocols/internal/infrastructure/postgres/mail_store.go +++ b/services/protocols/internal/infrastructure/postgres/mail_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sort" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -44,6 +45,27 @@ func NewMailStore(pool *pgxpool.Pool) *MailStore { return &MailStore{pool: pool} } +// lockFolders takes the row lock of each folder, in a fixed order. +// +// Every change updates counters on a folder row, and a move updates two. If one +// operation locked the destination folder and then the source, and another did +// the reverse, two moves in opposite directions deadlocked, and so did a move +// racing a flag change (it held the message and wanted the folder, the move held +// the folder and wanted the message). So the order is the same everywhere: +// folders first, lowest id first, then the message. +func lockFolders(ctx context.Context, tx pgx.Tx, ids ...string) error { + sort.Strings(ids) + for i, id := range ids { + if i > 0 && id == ids[i-1] { + continue + } + if _, err := tx.Exec(ctx, `SELECT 1 FROM folders WHERE id = $1 FOR UPDATE`, id); err != nil { + return fmt.Errorf("lock folder %s: %w", id, err) + } + } + return nil +} + // messageRow is the slice of a message every operation needs. type messageRow struct { ID uuid.UUID @@ -110,6 +132,9 @@ func (s *MailStore) SetFlags(ctx context.Context, folderID string, uid uint32, o } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, folderID); err != nil { + return false, err + } msg, err := loadMessage(ctx, tx, folderID, uid) if err != nil { return false, err @@ -239,6 +264,10 @@ func (s *MailStore) Expunge(ctx context.Context, folderID string, uids []uint32) } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, folderID); err != nil { + return 0, err + } + removed := 0 for _, uid := range uids { msg, err := loadMessage(ctx, tx, folderID, uid) @@ -288,14 +317,17 @@ func (s *MailStore) Move(ctx context.Context, sourceFolderID string, uid uint32, } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, sourceFolderID, destFolderID); err != nil { + return 0, err + } msg, err := loadMessage(ctx, tx, sourceFolderID, uid) if err != nil { return 0, err } var destUID int64 - if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) + if err := tx.QueryRow(ctx, `SELECT uid_next FROM folders WHERE id = $1`, destFolderID).Scan(&destUID); err != nil { + return 0, fmt.Errorf("read destination uid: %w", err) } unread := 0 @@ -353,6 +385,9 @@ func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUI } defer tx.Rollback(ctx) + if err := lockFolders(ctx, tx, destFolderID); err != nil { + return 0, err + } src, err := loadMessage(ctx, tx, sourceFolderID, sourceUID) if err != nil { return 0, err @@ -360,8 +395,8 @@ func (s *MailStore) copyOne(ctx context.Context, sourceFolderID string, sourceUI var destUID int64 var destMailboxID string - if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1 FOR UPDATE`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { - return 0, fmt.Errorf("lock destination folder: %w", err) + if err := tx.QueryRow(ctx, `SELECT uid_next, mailbox_id::text FROM folders WHERE id = $1`, destFolderID).Scan(&destUID, &destMailboxID); err != nil { + return 0, fmt.Errorf("read destination folder: %w", err) } unread := 0