panel,mail: fail closed on the rate-limit race, session create and app delete

The level-2 limiter counted stored plus in-flight messages and reserved its own slot in two critical sections, so SMTP sessions that overlapped could each take the last free slot; tryAdmit now does both under one lock. A session that cannot be written no longer yields a cookie the browser would carry while every request bounced to /login. Deleting an application clears its SASL account before its registry row, matching domain delete, so a saslpasswd2 failure leaves a retryable application rather than an account that still authenticates.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

internal/milter/inflight.go

@@ -36,6 +36,29 @@ type inflight struct {
 	m  map[string]map[*reservation]struct{}
 }
 
+// tryAdmit decides whether one more message may be sent for key and, if so,
+// claims its slot — both under a single lock. Counting and reserving in two
+// separate critical sections would let two SMTP sessions that reach MAIL FROM
+// at the same moment observe the same total and each take the last free slot,
+// which is exactly the overshoot the in-flight registry exists to prevent.
+//
+// stored is the count the send log already holds for the limit's window and max
+// is the ceiling; the caller supplies both because only it can query the store.
+// The returned total is what was measured, for the refusal log line.
+func (f *inflight) tryAdmit(key string, since time.Time, stored, max int64) (*reservation, int64, bool) {
+	if f == nil {
+		return nil, stored, stored < max // no in-flight accounting (tests)
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	total := stored + f.countLocked(key, since)
+	if total >= max {
+		return nil, total, false
+	}
+	return f.reserveLocked(key), total, true
+}
+
 // count returns how many reservations for key were taken within the limit's
 // window (at or after since), pruning any that outlived reservationTTL.
 func (f *inflight) count(key string, since time.Time) int64 {
@@ -45,6 +68,11 @@ func (f *inflight) count(key string, since time.Time) int64 {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
+	return f.countLocked(key, since)
+}
+
+// countLocked is count's body; the caller holds mu.
+func (f *inflight) countLocked(key string, since time.Time) int64 {
 	set := f.m[key]
 	cutoff := time.Now().Add(-reservationTTL)
 	var n int64
@@ -63,14 +91,8 @@ func (f *inflight) count(key string, since time.Time) int64 {
 	return n
 }
 
-// reserve claims a slot for key until the message is recorded or released.
-func (f *inflight) reserve(key string) *reservation {
-	if f == nil {
-		return nil
-	}
-	f.mu.Lock()
-	defer f.mu.Unlock()
-
+// reserveLocked is reserve's body; the caller holds mu.
+func (f *inflight) reserveLocked(key string) *reservation {
 	if f.m == nil {
 		f.m = make(map[string]map[*reservation]struct{})
 	}

internal/milter/milter_test.go

@@ -3,6 +3,7 @@ package milter
 import (
 	"errors"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -15,7 +16,10 @@ import (
 // swallows recorder errors and still accepts the message. By default it reports
 // no configured rate limit, so the level-2 check is inert unless a test sets
 // limits (see fakeRecorder fields).
+// mu guards the recorded slices so several sessions may drive one recorder
+// concurrently, as they do in the real server.
 type fakeRecorder struct {
+	mu       sync.Mutex
 	entries  []store.SendLogEntry
 	rejected []store.SendLogEntry
 	fail     bool
@@ -27,17 +31,25 @@ type fakeRecorder struct {
 	counts    map[string]int64
 	lookupErr error
 	countErr  error
+
+	// onCount, if set, runs inside CountMessages. It lets a test hold every
+	// racing session at the store lookup until they can all proceed together.
+	onCount func()
 }
 
 func (f *fakeRecorder) InsertQueued(e store.SendLogEntry) error {
 	if f.fail {
 		return errors.New("boom")
 	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.entries = append(f.entries, e)
 	return nil
 }
 
 func (f *fakeRecorder) InsertRejected(e store.SendLogEntry) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.rejected = append(f.rejected, e)
 	return nil
 }
@@ -57,6 +69,9 @@ func (f *fakeRecorder) CountMessages(scope, ref string, _ time.Time) (int64, err
 	if f.countErr != nil {
 		return 0, f.countErr
 	}
+	if f.onCount != nil {
+		f.onCount()
+	}
 	return f.counts[scope+"|"+ref], nil
 }
 
@@ -415,6 +430,58 @@ func TestRateLimitCountsInFlightMessages(t *testing.T) {
 	}
 }
 
+// The case above is sequential: the second session reads the stored count after
+// the first has already reserved. Here every session reads it first — the gate
+// holds them all inside the lookup — which is what concurrent SMTP connections
+// actually do. However many then race for the single free slot, exactly one may
+// pass. (TestTryAdmitHandsOutEachSlotOnce is the test that fails when counting
+// and reserving are not one step; this one pins the session-level behaviour.)
+func TestRateLimitAdmitsOnlyOneRacingSession(t *testing.T) {
+	rec := limitedRecorder(4) // one below the ceiling of 5
+	fl := &inflight{}
+	gate := make(chan struct{})
+
+	// Every session is held inside the stored-count lookup until all of them
+	// have read it, which is the state the race needs: none of them can see
+	// another's reservation, because none has been taken yet.
+	const racers = 32
+	var atCount, done sync.WaitGroup
+	atCount.Add(racers)
+	go func() { atCount.Wait(); close(gate) }()
+	rec.onCount = func() { atCount.Done(); <-gate }
+
+	responses := make([]milter.Response, racers)
+	for i := range racers {
+		done.Add(1)
+		go func() {
+			defer done.Done()
+			// Connect is skipped so every goroutine starts from the same point;
+			// the client IP is what Connect would have captured.
+			s := &session{rec: rec, flight: fl, clientIP: limitIP}
+			resp, err := s.MailFrom("a@example.com", mods(map[string]string{"auth_authen": "app1"}))
+			if err != nil {
+				resp = nil // reported as a missing Continue below
+			}
+			responses[i] = resp
+		}()
+	}
+	done.Wait()
+
+	admitted := 0
+	for _, resp := range responses {
+		if resp == milter.RespContinue {
+			admitted++
+		}
+	}
+	if admitted != 1 {
+		t.Fatalf("%d of %d racing sessions admitted, want exactly 1 — the last slot was handed out twice",
+			admitted, racers)
+	}
+	if n := fl.count(store.RateLimitScopeDomain+"|example.com", time.Now().Add(-time.Hour)); n != 1 {
+		t.Fatalf("in-flight reservations = %d, want 1", n)
+	}
+}
+
 // Once the message is recorded the stored count sees it, so its reservation
 // must be given back — otherwise it would be counted twice and the ceiling
 // would drift closed.
@@ -477,12 +544,55 @@ func TestRefusalDoesNotLeaveDomainReservation(t *testing.T) {
 	}
 }
 
+// The ceiling is handed out exactly max times however the sessions interleave.
+// Counting and reserving in two critical sections passes the sequential tests
+// above and still overshoots here, because between one session's count and its
+// reservation any number of others can pass the same check.
+func TestTryAdmitHandsOutEachSlotOnce(t *testing.T) {
+	const (
+		max     = 500
+		workers = 8
+	)
+	fl := &inflight{}
+	since := time.Now().Add(-time.Hour)
+	start := make(chan struct{})
+	admitted := make([]int, workers)
+	var wg sync.WaitGroup
+	for i := range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for {
+				_, _, ok := fl.tryAdmit("domain|example.com", since, 0, max)
+				if !ok {
+					return
+				}
+				admitted[i]++
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	total := 0
+	for _, n := range admitted {
+		total += n
+	}
+	if total != max {
+		t.Fatalf("admitted %d messages under a ceiling of %d", total, max)
+	}
+}
+
 // The in-flight count only covers the limit's own window: a reservation older
 // than it (a session stuck mid-DATA for longer than the window) must not be
 // counted against a window it no longer belongs to.
 func TestInflightIgnoresReservationsOutsideWindow(t *testing.T) {
 	fl := &inflight{}
-	r := fl.reserve("domain|example.com")
+	r, _, ok := fl.tryAdmit("domain|example.com", time.Now().Add(-time.Hour), 0, 1)
+	if !ok {
+		t.Fatal("tryAdmit refused the first message under a ceiling of 1")
+	}
 	r.at = time.Now().Add(-time.Minute)
 
 	if n := fl.count("domain|example.com", time.Now().Add(-time.Hour)); n != 1 {

internal/milter/ratelimit.go

@@ -54,22 +54,23 @@ func (s *session) overLimit() bool {
 }
 
 // enforceLimit counts recent messages for scope/ref and refuses when at or
-// above the ceiling. On admit it reserves an in-flight slot on the session.
+// above the ceiling. The stored count and the in-flight slots are weighed and
+// the admitted message's own slot is taken in one atomic step (tryAdmit), so
+// two sessions racing at MAIL FROM cannot both claim the last free slot.
 func (s *session) enforceLimit(scope, ref string, rl store.RateLimit) bool {
 	since := time.Now().Add(-time.Duration(rl.WindowSeconds) * time.Second)
-	n, err := s.rec.CountMessages(scope, ref, since)
+	stored, err := s.rec.CountMessages(scope, ref, since)
 	if err != nil {
 		log.Printf("journal-milter: rate-limit count %s %q: %v (fail-open)", scope, ref, err)
 		return false
 	}
-	key := scope + "|" + ref
-	n += s.flight.count(key, since)
-	if n >= int64(rl.MaxMessages) {
+	r, n, ok := s.flight.tryAdmit(scope+"|"+ref, since, stored, int64(rl.MaxMessages))
+	if !ok {
 		log.Printf("journal-milter: %s %q over limit: %d/%d in %ds from %s — refusing 4xx",
 			scope, ref, n, rl.MaxMessages, rl.WindowSeconds, s.clientIP)
 		return true
 	}
-	s.reserved = append(s.reserved, s.flight.reserve(key))
+	s.reserved = append(s.reserved, r)
 	return false
 }