feat: log-tailer offset persistence + in-flight L2 rate-limit accounting (code-review.md § Phase 3)

- logtail: persist the read position (offset + fingerprint of the log's
  first 512 bytes) in a new logtail_state table (migration 0003) and
  resume from it on start, so delivery lines written while the panel was
  down are parsed instead of skipped and their send-log rows no longer
  stay "queued" forever. Fingerprint mismatch (rotated/recreated while
  down) reads the file from the start — re-parsing is idempotent; a
  first-ever start with nothing stored still begins at end-of-file.
  Writes are throttled to one per 5s, forced on rotation and shutdown.

- milter: count messages that passed the level-2 check but have not
  reached the send log yet (internal/milter/inflight.go), so concurrent
  SMTP sessions cannot each spend the same last slot. A literal
  count+insert transaction, as the review suggested, is not possible:
  the count happens at MAIL FROM and the insert at end-of-message.
  Reservations are released after the insert, on ABORT, and after a
  10-minute TTL — a client that drops mid-transaction must not be able
  to hold a slot, since the limiter is fail-open by design.

Docs: architecture.md (log tailer, persistence, L2 counting),
security.md and roadmap.md (restart gap closed, container recreate
remains), CHANGELOG, progress.md, code-review.md.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

internal/milter/inflight.go

@@ -0,0 +1,99 @@
+package milter
+
+import (
+	"sync"
+	"time"
+)
+
+// reservationTTL bounds how long a message may stay reserved. A reservation is
+// released at end-of-message or on ABORT, but a client that simply drops the
+// connection after MAIL FROM produces neither callback (go-milter has no
+// connection-close hook), and a reservation that never expired would count
+// against the limit forever — a fail-closed drift this milter must not have.
+// The TTL is generously longer than any realistic DATA transfer, so a message
+// still being received is never dropped from the count.
+const reservationTTL = 10 * time.Minute
+
+// reservation is one message that passed the level-2 check and has not been
+// written to the send log yet.
+type reservation struct {
+	key string
+	at  time.Time
+}
+
+// inflight counts messages that are between the limit check (MAIL FROM) and the
+// send-log insert (end-of-message). The stored count alone cannot see them, so
+// without this several concurrent SMTP sessions each read the same pre-insert
+// count, each conclude they are under the ceiling, and the limit is overshot by
+// however many were in flight. Counting reservations closes that window without
+// writing placeholder rows the operator would see in the UI.
+//
+// One instance is shared by every session of the process, hence the mutex.
+// Methods tolerate a nil receiver so a session built without one (tests) simply
+// behaves as it did before.
+type inflight struct {
+	mu sync.Mutex
+	m  map[string]map[*reservation]struct{}
+}
+
+// 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 {
+	if f == nil {
+		return 0
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	set := f.m[key]
+	cutoff := time.Now().Add(-reservationTTL)
+	var n int64
+	for r := range set {
+		if r.at.Before(cutoff) {
+			delete(set, r)
+			continue
+		}
+		if !r.at.Before(since) {
+			n++
+		}
+	}
+	if len(set) == 0 {
+		delete(f.m, key)
+	}
+	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()
+
+	if f.m == nil {
+		f.m = make(map[string]map[*reservation]struct{})
+	}
+	if f.m[key] == nil {
+		f.m[key] = make(map[*reservation]struct{})
+	}
+	r := &reservation{key: key, at: time.Now()}
+	f.m[key][r] = struct{}{}
+	return r
+}
+
+// release drops a reservation, either because the message reached the send log
+// (where the stored count takes over) or because it never will.
+func (f *inflight) release(r *reservation) {
+	if f == nil || r == nil {
+		return
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	set := f.m[r.key]
+	delete(set, r)
+	if len(set) == 0 {
+		delete(f.m, r.key)
+	}
+}

internal/milter/milter.go

@@ -45,13 +45,18 @@ type Store interface {
 type session struct {
 	milter.NoOpMilter
 	rec Store
+	// flight is shared by every session of the process; it holds the messages
+	// that passed the level-2 check but are not in the send log yet. Nil is a
+	// valid zero value (no in-flight accounting).
+	flight *inflight
 
 	clientIP string // captured once per connection
 
-	login   string
-	from    string
-	rcpts   []string
-	subject string
+	login    string
+	from     string
+	rcpts    []string
+	subject  string
+	reserved []*reservation // level-2 slots held by the current message
 }
 
 // Connect captures the client IP, which comes from the addr parameter rather
@@ -72,6 +77,7 @@ func (s *session) Connect(host, family string, port uint16, addr net.IP, m *milt
 // tempfail before recipients are even offered (spec 7.4). Enforcement is
 // fail-open — see overLimit.
 func (s *session) MailFrom(from string, m *milter.Modifier) (milter.Response, error) {
+	s.releaseReservations() // a previous transaction that ended without EOM/ABORT
 	s.from = cleanAddress(from)
 	s.login = macro(m, "auth_authen")
 	s.rcpts = nil
@@ -127,9 +133,22 @@ func decodeSubject(v string) string {
 // rows are written. We accept (this milter is done) without ever rejecting.
 func (s *session) Body(m *milter.Modifier) (milter.Response, error) {
 	s.record(macro(m, "i"))
+	// The rows are in the send log now, so the stored count sees this message
+	// and its level-2 slots are no longer needed.
+	s.releaseReservations()
 	return milter.RespAccept, nil
 }
 
+// Abort ends the current transaction without an end-of-message (client RSET, or
+// Postfix rejecting the message for its own reasons). No send-log row will be
+// written, so the level-2 slots this message held must go back.
+func (s *session) Abort(m *milter.Modifier) error {
+	s.releaseReservations()
+	s.rcpts = nil
+	s.subject = ""
+	return nil
+}
+
 // macro reads a milter macro, tolerating Postfix's convention of wrapping
 // multi-character macro names in curly braces (e.g. {auth_authen}) while
 // single-character names (e.g. i) arrive bare. go-milter stores whatever name
@@ -191,8 +210,9 @@ func domainOf(addr string) string {
 // Serve runs the journal-milter on ln until ctx is cancelled. Each connection
 // gets a fresh session bound to rec. It returns nil on a clean shutdown.
 func Serve(ctx context.Context, ln net.Listener, rec Store) error {
+	flight := &inflight{} // shared: the level-2 window spans all connections
 	srv := &milter.Server{
-		NewMilter: func() milter.Milter { return &session{rec: rec} },
+		NewMilter: func() milter.Milter { return &session{rec: rec, flight: flight} },
 		Actions:   0,                // read-only: we make no message modifications
 		Protocol:  milter.OptNoBody, // the journal needs headers/EOM, not the body
 	}

internal/milter/milter_test.go

@@ -331,6 +331,130 @@ func TestRateLimitNoIPKeyDoesNotApply(t *testing.T) {
 	}
 }
 
+// mailFromIn is mailFrom with an explicit shared in-flight registry, so a test
+// can play several concurrent SMTP sessions of one process against each other.
+func mailFromIn(t *testing.T, rec Store, fl *inflight, ip, from, login string) (*session, milter.Response) {
+	t.Helper()
+	s := &session{rec: rec, flight: fl}
+	if _, err := s.Connect("h", "tcp4", 0, net.ParseIP(ip), mods(nil)); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	resp, err := s.MailFrom(from, mods(map[string]string{"auth_authen": login}))
+	if err != nil {
+		t.Fatalf("MailFrom: %v", err)
+	}
+	return s, resp
+}
+
+func limitedRecorder(count int64) *fakeRecorder {
+	return &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": count},
+	}
+}
+
+// Messages between MAIL FROM and end-of-message are not in the send log yet, so
+// counting the stored rows alone lets concurrent sessions each pass the same
+// check and overshoot the ceiling. The last free slot may only be taken once.
+func TestRateLimitCountsInFlightMessages(t *testing.T) {
+	rec := limitedRecorder(4) // one below the ceiling of 5
+	fl := &inflight{}
+
+	if _, resp := mailFromIn(t, rec, fl, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("first message = %v, want Continue (4/5 stored)", resp)
+	}
+	// Same window, nothing written yet: the first message holds the fifth slot.
+	if _, resp := mailFromIn(t, rec, fl, limitIP, "b@example.com", "app1"); resp != milter.RespTempFail {
+		t.Fatalf("concurrent message = %v, want TempFail (would overshoot)", resp)
+	}
+	if len(rec.rejected) != 1 {
+		t.Fatalf("want one rejected send-log row, got %+v", rec.rejected)
+	}
+}
+
+// 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.
+func TestReservationReleasedAtEndOfMessage(t *testing.T) {
+	rec := limitedRecorder(4)
+	fl := &inflight{}
+
+	s, resp := mailFromIn(t, rec, fl, limitIP, "a@example.com", "app1")
+	if resp != milter.RespContinue {
+		t.Fatalf("first message = %v, want Continue", resp)
+	}
+	if _, err := s.Body(mods(map[string]string{"i": "Q1"})); err != nil {
+		t.Fatalf("Body: %v", err)
+	}
+	if n := fl.count(store.RateLimitScopeDomain+"|example.com", time.Now().Add(-time.Hour)); n != 0 {
+		t.Fatalf("in-flight count after EOM = %d, want 0", n)
+	}
+}
+
+// A transaction the client abandons (RSET, or a Postfix-side rejection) never
+// reaches the send log, so its slot must not stay claimed.
+func TestReservationReleasedOnAbort(t *testing.T) {
+	rec := limitedRecorder(4)
+	fl := &inflight{}
+
+	s, resp := mailFromIn(t, rec, fl, limitIP, "a@example.com", "app1")
+	if resp != milter.RespContinue {
+		t.Fatalf("first message = %v, want Continue", resp)
+	}
+	if err := s.Abort(mods(nil)); err != nil {
+		t.Fatalf("Abort: %v", err)
+	}
+	if _, resp := mailFromIn(t, rec, fl, limitIP, "b@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("after abort = %v, want Continue (slot released)", resp)
+	}
+}
+
+// A refused message must not leave the slots it claimed for the limits checked
+// before the one that tripped, or every refusal would tighten the ceiling.
+func TestRefusalReleasesEarlierReservation(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+			store.RateLimitScopeApp + "|app1":           activeLimit(limitIP),
+		},
+		counts: map[string]int64{
+			store.RateLimitScopeDomain + "|example.com": 0, // domain: plenty of room
+			store.RateLimitScopeApp + "|app1":           5, // app: at the ceiling
+		},
+	}
+	fl := &inflight{}
+	if _, resp := mailFromIn(t, rec, fl, limitIP, "a@example.com", "app1"); resp != milter.RespTempFail {
+		t.Fatalf("app over limit = %v, want TempFail", resp)
+	}
+	if n := fl.count(store.RateLimitScopeDomain+"|example.com", time.Now().Add(-time.Hour)); n != 0 {
+		t.Fatalf("domain reservation left behind after refusal: %d", n)
+	}
+}
+
+// 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.at = time.Now().Add(-time.Minute)
+
+	if n := fl.count("domain|example.com", time.Now().Add(-time.Hour)); n != 1 {
+		t.Fatalf("count inside window = %d, want 1", n)
+	}
+	if n := fl.count("domain|example.com", time.Now().Add(-time.Second)); n != 0 {
+		t.Fatalf("count outside window = %d, want 0", n)
+	}
+	// Past the TTL the reservation is dropped even for a wide window, so a
+	// client that vanished after MAIL FROM cannot hold a slot forever.
+	r.at = time.Now().Add(-2 * reservationTTL)
+	if n := fl.count("domain|example.com", time.Now().Add(-3*reservationTTL)); n != 0 {
+		t.Fatalf("expired reservation still counted: %d", n)
+	}
+}
+
 func TestDomainOf(t *testing.T) {
 	cases := map[string]string{
 		"user@Example.COM": "example.com",

internal/milter/ratelimit.go

@@ -17,6 +17,10 @@ import (
 // can never block mail — Postfix's level-1 anvil limit (spec 5) remains the
 // backstop, and it does not depend on this milter at all. Only a clean count at
 // or above a configured ceiling returns true.
+//
+// A message that passes reserves a slot per applicable limit, released once it
+// reaches the send log (or is abandoned) — see inflight for why the stored
+// count alone is not enough.
 func (s *session) overLimit() bool {
 	if s.clientIP == "" {
 		return false // no client IP to key on; level-2 does not apply
@@ -25,6 +29,7 @@ func (s *session) overLimit() bool {
 		{store.RateLimitScopeDomain, domainOf(s.from)},
 		{store.RateLimitScopeApp, s.login},
 	}
+	var taken []*reservation
 	for _, c := range checks {
 		if c.ref == "" {
 			continue
@@ -45,15 +50,34 @@ func (s *session) overLimit() bool {
 			log.Printf("journal-milter: rate-limit count %s %q: %v (fail-open)", c.scope, c.ref, err)
 			continue
 		}
+		key := c.scope + "|" + c.ref
+		n += s.flight.count(key, since)
 		if n >= int64(rl.MaxMessages) {
 			log.Printf("journal-milter: %s %q over limit: %d/%d in %ds from %s — refusing 4xx",
 				c.scope, c.ref, n, rl.MaxMessages, rl.WindowSeconds, s.clientIP)
+			// The message is refused, so the slots claimed for the limits
+			// checked before this one must not stay claimed.
+			for _, r := range taken {
+				s.flight.release(r)
+			}
 			return true
 		}
+		taken = append(taken, s.flight.reserve(key))
 	}
+	s.reserved = append(s.reserved, taken...)
 	return false
 }
 
+// releaseReservations gives back every slot this message holds. It runs once
+// the message is in the send log (where the stored count sees it), and whenever
+// the transaction ends without getting there.
+func (s *session) releaseReservations() {
+	for _, r := range s.reserved {
+		s.flight.release(r)
+	}
+	s.reserved = nil
+}
+
 // recordRejected writes a send-log row for a message refused by a level-2 limit
 // (spec 7.4, "опционально фиксирует ... для видимости в UI"), so the rejection
 // shows up in the monitoring screen. Only MAIL-stage fields are known; the write