diff --git a/internal/httpapi/gmail_handlers.go b/internal/httpapi/gmail_handlers.go index 3ef1b46..74dc370 100644 --- a/internal/httpapi/gmail_handlers.go +++ b/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 } diff --git a/internal/httpapi/mailbox_handlers.go b/internal/httpapi/mailbox_handlers.go index 8c1b135..8baaeab 100644 --- a/internal/httpapi/mailbox_handlers.go +++ b/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 +} diff --git a/internal/httpapi/panel_test.go b/internal/httpapi/panel_test.go index d28dff2..cf6dc06 100644 --- a/internal/httpapi/panel_test.go +++ b/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() +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index d3913cf..e6f1c38 100644 --- a/internal/httpapi/server.go +++ b/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