fix(logtail): keep mail.log in /data and reconcile stuck rows (v1.x closure phase 2)

Move the delivery log from the ephemeral /var/log to /data/log/mail.log so
the lines that resolve a queued send-log row survive a container recreate.
postlogd writes it as postfix, the panel reads it through the selfpost group
(dir 2750, file 0640, normalised every start); backups exclude log/.

Close the residual gap with a queue sweep: rows queued for over two minutes
whose id postqueue -p no longer lists are marked bounced. The sweep waits
until the tailer has read the log to its end and does nothing when the queue
cannot be listed, so a message in flight is never touched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
mixeme
2026-08-08 11:35:27 +03:00
parent 7e4ecf1191
commit 1f548dfc7a
25 changed files with 549 additions and 72 deletions
+102 -4
View File
@@ -16,16 +16,20 @@ import (
"os"
"regexp"
"strings"
"sync"
"time"
"github.com/mixeme/selfpost/internal/postfix"
"github.com/mixeme/selfpost/internal/store"
)
// StatusStore is the slice of the store the log-tailer needs: advancing
// delivery statuses, pruning the retention window, and remembering how far into
// mail.log it has read. *store.Store satisfies it.
// delivery statuses, finding the rows still waiting for one, pruning the
// retention window, and remembering how far into mail.log it has read.
// *store.Store satisfies it.
type StatusStore interface {
UpdateStatus(queueID, recipient, status string) (int64, error)
ListQueuedOlderThan(cutoff time.Time) ([]store.QueuedDelivery, error)
DeleteSendLogBefore(cutoff time.Time) (int64, error)
LogtailState(path string) (store.LogtailState, bool, error)
SaveLogtailState(path string, st store.LogtailState) error
@@ -35,6 +39,10 @@ type StatusStore interface {
// is a var so tests can shorten it.
var pollInterval = time.Second
// queueIDs lists the messages Postfix currently holds, for the reconcile sweep.
// It is a var so tests can answer without a running Postfix.
var queueIDs = postfix.QueueIDs
const (
// retentionInterval is how often the retention sweep runs (also once at
// startup). The window itself is configurable; the cadence need not be.
@@ -42,6 +50,13 @@ const (
// defaultRetentionDays applies when the configured value is unset/invalid
// (guide § Environment variables: SEND_LOG_RETENTION_DAYS).
defaultRetentionDays = 90
// reconcileInterval is how often the sweep compares stuck rows against the
// Postfix queue, and reconcileGrace how long a row is left alone first.
// The grace covers the ordinary lag between the milter writing the row and
// Postfix logging the result — seconds, generously rounded up — so a
// message merely in flight is never touched.
reconcileInterval = 5 * time.Minute
reconcileGrace = 2 * time.Minute
)
// deliveryRe matches a Postfix delivery line and captures queue-id, recipient
@@ -89,7 +104,14 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
func Run(ctx context.Context, path string, st StatusStore, retentionDays int) error {
go retentionLoop(ctx, st, retentionDays)
return follow(ctx, path, &tracker{st: st, path: path}, func(line string) {
// The reconcile sweep must not run against a backlog the tailer has not
// read yet: on a restart the log holds the very lines that resolve the rows
// the sweep would otherwise close. follow() closes this once it has read to
// end-of-file for the first time.
caughtUp := make(chan struct{})
go reconcileLoop(ctx, st, caughtUp)
return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, func(line string) {
queueID, recipient, status, ok := parseDelivery(line)
if !ok {
return
@@ -100,6 +122,75 @@ func Run(ctx context.Context, path string, st StatusStore, retentionDays int) er
})
}
// reconcileLoop periodically closes send-log rows Postfix has stopped working
// on (architecture.md § Log tailer). It starts only once the tailer has caught
// up with the log, and then leaves the first sweep a full interval away, so a
// restart resolves rows from the log — the accurate source — before the sweep
// gets to guess at whatever the log could not explain.
func reconcileLoop(ctx context.Context, st StatusStore, caughtUp <-chan struct{}) {
select {
case <-ctx.Done():
return
case <-caughtUp:
}
t := time.NewTicker(reconcileInterval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
reconcile(st, time.Now().UTC().Add(-reconcileGrace))
}
}
}
// reconcile marks as bounced every row still "queued" from before cutoff whose
// message Postfix no longer holds.
//
// A row reaches this state only when its delivery lines are gone for good — the
// log rotated past its fourteen files while the panel was down, or was deleted
// — since the log itself now outlives the container. Postfix having dropped the
// message means it will never report anything more about it, so the row can
// only be closed on an assumption; it is closed as a failure rather than a
// success because a delivery the panel cannot evidence must not be shown as
// one. Rows whose message is still in the queue, and every row when the queue
// cannot be listed at all, are left exactly as they are.
func reconcile(st StatusStore, cutoff time.Time) {
rows, err := st.ListQueuedOlderThan(cutoff)
if err != nil {
log.Printf("log-tailer: reconcile: list queued rows: %v", err)
return
}
if len(rows) == 0 {
return
}
held, err := queueIDs()
if err != nil {
// No listing is no information: closing rows now would be a guess made
// against nothing.
log.Printf("log-tailer: reconcile: read postfix queue: %v", err)
return
}
var closed int
for _, row := range rows {
if _, still := held[row.QueueID]; still {
continue
}
if _, err := st.UpdateStatus(row.QueueID, row.To, store.StatusBounced); err != nil {
log.Printf("log-tailer: reconcile: close %s/%s: %v", row.QueueID, row.To, err)
continue
}
closed++
}
if closed > 0 {
log.Printf("log-tailer: reconcile: closed %d row(s) Postfix no longer holds and never reported", closed)
}
}
// retentionLoop prunes expired send-log rows immediately and then periodically.
func retentionLoop(ctx context.Context, st StatusStore, retentionDays int) {
if retentionDays <= 0 {
@@ -323,7 +414,11 @@ func SplitTimestamp(line string) (stamp, rest string) {
// start of a file that changed while the panel was down, or end-of-file on a
// first ever run); it reopens the file when it is rotated (inode change from
// logrotate's create, or truncation from copytruncate) so nothing is missed.
func follow(ctx context.Context, path string, tr *tracker, handle func(string)) error {
//
// caughtUp is closed after the first read that reaches end-of-file, which is
// the point where every line the panel missed while it was down has been
// handled.
func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{}, handle func(string)) error {
var (
f *os.File
r *bufio.Reader
@@ -397,6 +492,8 @@ func follow(ctx context.Context, path string, tr *tracker, handle func(string))
return pos - int64(len(pending))
}
var once sync.Once
ticker := time.NewTicker(pollInterval)
defer ticker.Stop()
for {
@@ -406,6 +503,7 @@ func follow(ctx context.Context, path string, tr *tracker, handle func(string))
return nil
case <-ticker.C:
drain()
once.Do(func() { close(caughtUp) })
ni, err := os.Stat(path)
if err != nil {
continue // file briefly gone mid-rotation; try again next tick