fix(mailboxes): to kørsler af samme postkasse må ikke læse den samme markør

Fyrre kopier af én mail ankom i samme sekund. Én kørsel kan højst lave femogtyve,
så der var to i gang: et tryk på "Hent nu" mens ti-minutters-fejningen allerede
var inde i den samme postkasse. Begge læste markøren som den stod, begge hentede
de samme beskeder, begge lavede opgaver ud af dem.

Markør-rettelsen fra i går var altså rigtig, men ikke nok — den virker, når der
kun er én læser. Rækken læses nu igen under låsen, og låsen holdes hele vejen
igennem: slap man den, før markøren var skrevet, ville hullet stå åbent præcis
der, hvor det skulle lukkes.

Et kort pr. postkasse frem for én lås. To menneskers postkasser har intet at sige
hinanden, og én fælles lås ville gøre én langsom vært til alles problem.

Prøven kører to kørsler oven i hinanden med vilje og tæller, hvor mange der er
inde ad gangen — en lås er kun værd at have, hvis noget prøver døren. Den holder
også øje med, at to forskellige postkasser ikke venter på hinanden.

Co-Authored-By: Claude Opus 5 <[email protected]>

internal/httpapi/gmail_handlers.go

@@ -256,6 +256,16 @@ func (s *Server) SyncGmail(ctx context.Context, user *store.User) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+	if box != nil {
+		// The same guard as the mailboxes read over IMAP, for the same reason: a
+		// press of "Fetch now" during the sweep would otherwise make two runs from
+		// one marker.
+		unlock := s.lockMailbox(box.ID)
+		defer unlock()
+		if fresh, err := s.db.MailboxOfKind(ctx, user.ID, "gmail"); err == nil && fresh != nil {
+			box = fresh
+		}
+	}
 	if box == nil || box.RefreshToken == "" {
 		return 0, nil // not connected; nothing to do and nothing to complain about
 	}

internal/httpapi/mailbox_handlers.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/go-chi/chi/v5"
@@ -139,6 +140,18 @@ func (s *Server) handleSyncMailbox(w http.ResponseWriter, r *http.Request) {
 // Exported because the background job calls it too, on the same terms and without
 // a budget: nobody is waiting there.
 func (s *Server) SyncMailbox(ctx context.Context, user *store.User, m *store.Mailbox) (int, error) {
+	// One reader of this mailbox at a time. Held for the whole run, marker and all:
+	// releasing before the marker is written would leave exactly the gap this is
+	// here to close.
+	unlock := s.lockMailbox(m.ID)
+	defer unlock()
+
+	// Re-read under the lock. The row that came in may have been fetched before the
+	// run that just finished moved the marker.
+	if fresh, err := s.db.Mailbox(ctx, m.UserID, m.ID); err == nil && fresh != nil {
+		m = fresh
+	}
+
 	client, err := imap.Dial(imap.Account{
 		Host: m.Host, Username: m.Username, Password: m.Password, Folder: m.Folder,
 	})
@@ -217,3 +230,11 @@ func (s *Server) SyncMailbox(ctx context.Context, user *store.User, m *store.Mai
 	}
 	return created, nil
 }
+
+// lockMailbox serialises one mailbox against itself and returns the release.
+func (s *Server) lockMailbox(id string) func() {
+	value, _ := s.syncing.LoadOrStore(id, &sync.Mutex{})
+	mu := value.(*sync.Mutex)
+	mu.Lock()
+	return mu.Unlock
+}

internal/httpapi/panel_test.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -82,3 +83,69 @@ func TestAPanelThatRefusesSaysSoAsJSON(t *testing.T) {
 		t.Errorf("the reply does not say what the panel answered: %v", body)
 	}
 }
+
+// Forty copies of one mail arrived in a single second because two runs read the
+// same marker before either wrote a new one — somebody pressing "Fetch now" while
+// the ten-minute sweep was already inside the same mailbox.
+//
+// The lock is what stops it, and a lock is only worth having if something tries
+// the door: this runs the two concurrently on purpose.
+func TestTwoRunsOfOneMailboxDoNotOverlap(t *testing.T) {
+	// A bare server: the lock needs nothing but its own zero value, and building a
+	// database and a session around it would only make the test slower to read.
+	srv := &Server{}
+
+	var inside, most int
+	var mu sync.Mutex
+	release := make(chan struct{})
+
+	// Stands in for a run: takes the lock, notes how many are inside, and holds it
+	// until told to let go.
+	run := func(done chan<- struct{}) {
+		unlock := srv.lockMailbox("box-1")
+		mu.Lock()
+		inside++
+		if inside > most {
+			most = inside
+		}
+		mu.Unlock()
+
+		<-release
+
+		mu.Lock()
+		inside--
+		mu.Unlock()
+		unlock()
+		done <- struct{}{}
+	}
+
+	first, second := make(chan struct{}), make(chan struct{})
+	go run(first)
+	go run(second)
+
+	// Long enough that both would be inside if nothing kept them apart.
+	time.Sleep(100 * time.Millisecond)
+	close(release)
+	<-first
+	<-second
+
+	if most != 1 {
+		t.Errorf("%d runs were inside the same mailbox at once; that is how one mail "+
+			"becomes forty tasks", most)
+	}
+
+	// And two different mailboxes must not wait for each other: one slow host is
+	// not everybody's problem.
+	a := srv.lockMailbox("box-a")
+	done := make(chan struct{})
+	go func() {
+		srv.lockMailbox("box-b")()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Error("a second mailbox waited for the first")
+	}
+	a()
+}

internal/httpapi/server.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/go-chi/chi/v5"
@@ -40,6 +41,18 @@ type Server struct {
 	mcp     *mcp.Server
 	updates *update.Checker
 
+	// syncing serialises reads of one mailbox against itself. Two runs at once —
+	// somebody pressing "Fetch now" while the ten-minute sweep is in the middle of
+	// the same mailbox — both start from the marker as it was, both fetch the same
+	// messages, and both make tasks out of them. That is how forty copies of one
+	// mail arrived in a single second.
+	//
+	// A map of mutexes rather than one lock: two people's mailboxes have nothing to
+	// say to each other, and a single lock would make one slow host everybody's
+	// problem. Never cleaned out, because the key is a mailbox id and there are as
+	// many of those as there are mailboxes.
+	syncing sync.Map
+
 	// Nil when the base URL is not something an authenticator will accept — a bare
 	// hostname with no scheme, say. The endpoints answer 503 rather than the
 	// process refusing to start: passkeys are one way in among three, and an