mix
a92d583053
- 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>
100 lines
2.7 KiB
Go
100 lines
2.7 KiB
Go
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)
|
|
}
|
|
}
|