refactor(mail): one send path for Sent, a fixed lock order, and concurrency tests

- the webmail's Send no longer files its own Sent copy: it asks Submit to
  (FileSentCopy), the same path SMTP clients use, so the rule exists once
- every operation now locks the folders first (lowest id first) and the message
  after. Opposite moves, and a move racing a flag change, deadlocked (SQLSTATE
  40P01) because they took those locks in different orders; the new test
  reproduces it (10-19 s of retries, errors) and passes in ~1 s with the order
- TestOneMailboxEverywhere drives SMTP, IMAP and the webmail on one mailbox,
  mixing them, and checks after each step that every view agrees with the others
  and with the rows. It fails on the previous code and passes now.

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

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