mix
ed0a786739
- 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>
55 lines
1.7 KiB
Go
55 lines
1.7 KiB
Go
package store
|
|
|
|
import (
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
// LogtailState is the persisted read position of the mail.log tailer: how far
|
|
// into the log it had got, and a fingerprint of the file that offset refers to.
|
|
// It lets the tailer resume after a panel restart instead of jumping to
|
|
// end-of-file and losing the delivery lines written while it was down.
|
|
type LogtailState struct {
|
|
Fingerprint string
|
|
Offset int64
|
|
}
|
|
|
|
// LogtailState returns the stored read position for path. ok is false the first
|
|
// time a path is followed (nothing persisted yet), which the tailer treats as
|
|
// "start at the end".
|
|
func (s *Store) LogtailState(path string) (LogtailState, bool, error) {
|
|
var st LogtailState
|
|
err := s.db.QueryRow(
|
|
`SELECT fingerprint, read_offset FROM logtail_state WHERE path = ?`,
|
|
path,
|
|
).Scan(&st.Fingerprint, &st.Offset)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return LogtailState{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return LogtailState{}, false, fmt.Errorf("read logtail state for %q: %w", path, err)
|
|
}
|
|
return st, true, nil
|
|
}
|
|
|
|
// SaveLogtailState records the tailer's read position for path, replacing any
|
|
// previous one. It is called on a timer while tailing, so it is a single small
|
|
// upsert rather than a transaction.
|
|
func (s *Store) SaveLogtailState(path string, st LogtailState) error {
|
|
_, err := s.db.Exec(
|
|
`INSERT INTO logtail_state (path, fingerprint, read_offset, updated_at)
|
|
VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(path) DO UPDATE SET
|
|
fingerprint = excluded.fingerprint,
|
|
read_offset = excluded.read_offset,
|
|
updated_at = excluded.updated_at`,
|
|
path, st.Fingerprint, st.Offset, time.Now().UTC().Format(time.RFC3339),
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("save logtail state for %q: %w", path, err)
|
|
}
|
|
return nil
|
|
}
|