logtail: a refusal the panel did not get to write is written from mail.log (stage 2b)
test / test (push) Canceled after 0s
test / test (push) Canceled after 0s
The sender check answers 550 5.7.1 and writes the refused row through a
buffer a moment later; a panel stopped in between (a rollout, a kill) left
an application refused and a log that did not know it. A refusal at MAIL
FROM needs nothing: the journal milter writes its row before it answers.
- logtail.refusalLines remembers each queue id's login from smtpd's
"<id>: client=..., sasl_username=<login>" (4096 ids at most) and, on
cleanup's "<id>: milter-reject: END-OF-MESSAGE ...: 5.7.1 From: <line> -
<why>; from=<...> to=<...>", waits 30 s and hands it to
store.RecoverRefusal - an INSERT ... WHERE NOT EXISTS a refused row of that
queue id within two minutes: the row the panel wrote, or one recovered
before, means nothing is written. It is read from the saved offset on, so
a kill is covered as well as a clean stop - which a flush of the buffer at
shutdown would not.
- Migration 0009 (not deployed yet) gains send_log.recovered; the row is
written with the line's time, the login's domain id, reason "from", the
From: line as the check printed it (to the last " - " before ";").
- The message page: a fact "Written: from mail.log after a restart: the
panel stopped before it wrote the refusal".
Tests: store (not written when the panel wrote it; written once however
often); logtail (the login from client=, the From: line with a " - " in it,
nothing before the wait is over); view (the fact). e2e
refusal_recovered_from_the_log: a From: refusal, the panel stopped, its row
deleted and the tailer's saved place moved before the message's first line
(python3 in the container, as the panel's user - a kill cannot be timed to
that moment from outside); started again, the row is back, recovered, once.
Also fixed: handlers TestMessagePageOfARefusedMessage still expected the
old sentence; it failed on the server after e166a2e, whose run showed only
the schema failures - now it wants the new one.
schema-migrations row 9, CHANGELOG [Unreleased], audit plan line ticked,
roadmap 26/62.
Evidence on the test server for this tree but the handler-test fix: make e2e
exit 0 (the new case 37 s); go test 22/23 ok, the 23rd that test - fixed,
passes locally; the full run follows before the push.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -286,6 +286,7 @@ func (c *captureStore) AddTailerBlind(time.Time, time.Duration) error { return n
|
||||
func (c *captureStore) AttachRefusalLine(string, string, time.Time, string) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
func (c *captureStore) RecoverRefusal(store.SendLogEntry, time.Time) (bool, error) { return false, nil }
|
||||
func (c *captureStore) CountQueuedOlderThan(time.Time) (int64, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
@@ -3,7 +3,10 @@ package logtail
|
||||
import (
|
||||
"log"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/mixeme/selfpost/internal/store"
|
||||
)
|
||||
|
||||
// Postfix's line about a message the journal milter refused at MAIL FROM - a
|
||||
@@ -24,23 +27,65 @@ const refusalLineWait = 30 * time.Second
|
||||
// RefusalStore is where the lines go. *store.Store satisfies it.
|
||||
type RefusalStore interface {
|
||||
AttachRefusalLine(login, from string, at time.Time, line string) (bool, error)
|
||||
RecoverRefusal(e store.SendLogEntry, at time.Time) (bool, error)
|
||||
}
|
||||
|
||||
// fromRefusalRe is cleanup's line about a message the sender check refused
|
||||
// for its From: line, as the stand writes it:
|
||||
//
|
||||
// postfix/cleanup[1339]: E543D3C75DA0: milter-reject: END-OF-MESSAGE from unknown[10.77.0.1]: 5.7.1 From: director@other.e2e.test - not an address this application sends as; from=<alerts@sender.e2e.test> to=<rcpt@sink.e2e.test> proto=ESMTP helo=<e2e-client.e2e.test>
|
||||
//
|
||||
// The From: line as the check printed it runs to the last " - " before the
|
||||
// semicolon; the reason after it holds neither.
|
||||
var fromRefusalRe = regexp.MustCompile(`^\S+ postfix/(?:[a-z0-9-]+/)?cleanup\[\d+\]: ([0-9A-Za-z]+): milter-reject: END-OF-MESSAGE from \S+: 5\.7\.1 From: (.*) - ([^;]*); from=<([^>]*)> to=<([^>]*)>`)
|
||||
|
||||
// clientLoginRe is smtpd's line naming the login a queue id was submitted
|
||||
// under: "<id>: client=host[addr], sasl_method=PLAIN, sasl_username=<login>".
|
||||
var clientLoginRe = regexp.MustCompile(`^\S+ postfix/(?:([a-z0-9-]+)/)?smtpd\[\d+\]: ([0-9A-Za-z]+): client=.*?, sasl_username=(\S+)`)
|
||||
|
||||
// refusalLogins bounds the queue ids whose login is remembered.
|
||||
const refusalLogins = 4096
|
||||
|
||||
type pendingRefusal struct {
|
||||
login, from, line string
|
||||
at, until time.Time
|
||||
}
|
||||
|
||||
// pendingRecovery is a From: refusal whose row is looked for once the panel
|
||||
// has had the time to write it.
|
||||
type pendingRecovery struct {
|
||||
e store.SendLogEntry
|
||||
at, until time.Time
|
||||
}
|
||||
|
||||
// refusalLines puts each refusal line on its row, and keeps for a while the
|
||||
// lines whose row is not there yet.
|
||||
// lines whose row is not there yet. It also writes, from cleanup's line, the
|
||||
// row of a From: refusal the panel did not get to write.
|
||||
type refusalLines struct {
|
||||
st RefusalStore
|
||||
pending []pendingRefusal
|
||||
st RefusalStore
|
||||
pending []pendingRefusal
|
||||
recover []pendingRecovery
|
||||
logins map[string]string // queue id -> login, from smtpd's client= line
|
||||
loginKeys []string
|
||||
}
|
||||
|
||||
// handle looks at one mail.log line.
|
||||
func (r *refusalLines) handle(line string, now time.Time) {
|
||||
at, rest := lineTime(line, now)
|
||||
if m := clientLoginRe.FindStringSubmatch(rest); m != nil && m[1] != "inbound" {
|
||||
r.remember(m[2], m[3])
|
||||
return
|
||||
}
|
||||
if m := fromRefusalRe.FindStringSubmatch(rest); m != nil {
|
||||
r.recover = append(r.recover, pendingRecovery{
|
||||
e: store.SendLogEntry{
|
||||
QueueID: m[1], AppLogin: r.logins[m[1]], From: m[4], To: m[5], Domain: domainOf(m[4]),
|
||||
Reason: store.ReasonFrom, HeaderFrom: m[2],
|
||||
},
|
||||
at: at, until: now.Add(refusalLineWait),
|
||||
})
|
||||
return
|
||||
}
|
||||
m := refusalLineRe.FindStringSubmatch(rest)
|
||||
if m == nil || m[1] == "inbound" {
|
||||
return
|
||||
@@ -52,7 +97,8 @@ func (r *refusalLines) handle(line string, now time.Time) {
|
||||
}
|
||||
|
||||
// retry tries the waiting lines again and lets go of those that waited long
|
||||
// enough.
|
||||
// enough; and writes, once their wait is over, the rows of From: refusals the
|
||||
// panel did not write.
|
||||
func (r *refusalLines) retry(now time.Time) {
|
||||
kept := r.pending[:0]
|
||||
for _, p := range r.pending {
|
||||
@@ -62,6 +108,45 @@ func (r *refusalLines) retry(now time.Time) {
|
||||
kept = append(kept, p)
|
||||
}
|
||||
r.pending = kept
|
||||
|
||||
waiting := r.recover[:0]
|
||||
for _, p := range r.recover {
|
||||
if !now.After(p.until) {
|
||||
waiting = append(waiting, p)
|
||||
continue
|
||||
}
|
||||
wrote, err := r.st.RecoverRefusal(p.e, p.at)
|
||||
if err != nil {
|
||||
log.Printf("log-tailer: recover refusal: %v", err)
|
||||
} else if wrote {
|
||||
log.Printf("log-tailer: wrote the refusal of %s (%s) from mail.log: the panel had not written it", p.e.QueueID, p.e.From)
|
||||
}
|
||||
}
|
||||
r.recover = waiting
|
||||
}
|
||||
|
||||
// remember keeps the login a queue id was submitted under, the oldest going
|
||||
// first beyond refusalLogins.
|
||||
func (r *refusalLines) remember(queueID, login string) {
|
||||
if r.logins == nil {
|
||||
r.logins = make(map[string]string)
|
||||
}
|
||||
if _, ok := r.logins[queueID]; !ok {
|
||||
r.loginKeys = append(r.loginKeys, queueID)
|
||||
if len(r.loginKeys) > refusalLogins {
|
||||
delete(r.logins, r.loginKeys[0])
|
||||
r.loginKeys = r.loginKeys[1:]
|
||||
}
|
||||
}
|
||||
r.logins[queueID] = login
|
||||
}
|
||||
|
||||
// domainOf is the domain of an address, lower-cased; "" without one.
|
||||
func domainOf(addr string) string {
|
||||
if i := strings.LastIndexByte(addr, '@'); i >= 0 {
|
||||
return strings.ToLower(addr[i+1:])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (r *refusalLines) attach(p pendingRefusal) bool {
|
||||
|
||||
@@ -3,12 +3,20 @@ package logtail
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mixeme/selfpost/internal/store"
|
||||
)
|
||||
|
||||
type refusalProbe struct {
|
||||
rows int // rows waiting for a line
|
||||
lines []string
|
||||
who []string
|
||||
rows int // rows waiting for a line
|
||||
lines []string
|
||||
who []string
|
||||
recovered []store.SendLogEntry
|
||||
}
|
||||
|
||||
func (r *refusalProbe) RecoverRefusal(e store.SendLogEntry, _ time.Time) (bool, error) {
|
||||
r.recovered = append(r.recovered, e)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (r *refusalProbe) AttachRefusalLine(login, from string, _ time.Time, line string) (bool, error) {
|
||||
@@ -56,3 +64,29 @@ func TestRefusalLines(t *testing.T) {
|
||||
t.Fatal("a line kept waiting past its time")
|
||||
}
|
||||
}
|
||||
|
||||
// cleanup's line about a From: refusal is the row's, written by the panel; once
|
||||
// the panel has had the time to, the line is handed to RecoverRefusal, which
|
||||
// writes the row only if it is missing - with the login smtpd named for the
|
||||
// queue id, the sender, the recipient and the From: line as printed.
|
||||
func TestFromRefusalIsRecoveredAfterTheWait(t *testing.T) {
|
||||
now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
|
||||
st := &refusalProbe{}
|
||||
r := &refusalLines{st: st}
|
||||
r.handle(`2026-10-14T12:00:00+00:00 mail postfix/smtpd[11]: E543D3C75DA0: client=unknown[10.77.0.1], sasl_method=PLAIN, sasl_username=app1`, now)
|
||||
r.handle(`2026-10-14T12:00:01+00:00 mail postfix/cleanup[13]: E543D3C75DA0: milter-reject: END-OF-MESSAGE from unknown[10.77.0.1]: 5.7.1 From: "The - Shop" <director@other.e2e.test> - it names another sender of this server; from=<alerts@sender.e2e.test> to=<rcpt@sink.e2e.test> proto=ESMTP helo=<x>`, now)
|
||||
r.retry(now.Add(time.Second))
|
||||
if len(st.recovered) != 0 {
|
||||
t.Fatal("recovered before the panel had the time to write the row")
|
||||
}
|
||||
r.retry(now.Add(refusalLineWait + time.Second))
|
||||
if len(st.recovered) != 1 {
|
||||
t.Fatalf("recovered %d, want 1", len(st.recovered))
|
||||
}
|
||||
e := st.recovered[0]
|
||||
want := store.SendLogEntry{QueueID: "E543D3C75DA0", AppLogin: "app1", From: "alerts@sender.e2e.test", To: "rcpt@sink.e2e.test",
|
||||
Domain: "sender.e2e.test", Reason: store.ReasonFrom, HeaderFrom: `"The - Shop" <director@other.e2e.test>`}
|
||||
if e != want {
|
||||
t.Fatalf("recovered %+v, want %+v", e, want)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -102,3 +102,7 @@ func TestRetentionLoopUsesUpdatedValue(t *testing.T) {
|
||||
func (s *retentionProbeStore) AttachRefusalLine(string, string, time.Time, string) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *retentionProbeStore) RecoverRefusal(store.SendLogEntry, time.Time) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user