Compare commits

..
2 Commits
Author SHA1 Message Date
mixandClaude Opus 5.5 c452606175 logtail: the deferred counter remembers a place in the log, not a second (stage 2b)
test / test (push) Canceled after 0s
The counter passed over, while catching up after a start, every warning no
later than the last one counted - by time; mail.log stamps whole seconds,
so a second warning of that second, later in the file, was lost.

- The tailer gives each line its place (Place: the file's fingerprint and
  the offset just past the line): follow tracks the bytes the reader hands
  over, readRotated likewise in mail.log.1. Run hands tr.line to the
  counter.
- store.AddSenderCheckDeferred(at, n, place) counts, moves the
  last-deferral mark (never back - an upsert WHERE) and saves the place
  (settings sender_check_counted_place) in one transaction.
- Catching up, a warning in the file of the last one counted is passed over
  by place (end <= counted end); in another file - a rotated log read again
  - by time, as before. The fingerprint carries the inode, so the place
  stays true across a rotation.

Tests: two warnings of one second, an unclean stop between them, all lines
read again - each counted once; the store saves the place with the count
and keeps the latest deferral time. Existing counter tests pass a zero
place (time rule).

CHANGELOG [Unreleased], audit plan line ticked, roadmap 24/62.
Evidence: go vet ./... clean; go test ./internal/store, ./internal/logtail ok
(but for the Windows-only test); the server run follows before the push.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 16:28:13 +03:00
mixandClaude Opus 5.5 4678015bbd logtail: a log rotated while the panel was down is read to its end (stage 2b)
At start, a saved offset whose fingerprint was not the current file's meant
"start on the new file": what Postfix wrote between the panel's stop and the
rotation was never read - delivery statuses, stored lines, refusals, inbound
journal rows, the deferred counter.

- tracker.resume marks the case (behind, prev); follow calls readRotated
  before the new file: it opens mail.log.1 (delaycompress leaves it plain),
  checks by fingerprint that it is the file the place belongs to, reads it
  from the saved offset to its end through handle (catching up: the
  once-protections apply) and flushes the stored lines.
- Not found, not that file, or the offset past its end (rotated twice):
  logged, with the saved place's time - from when nothing was read.
- The fingerprint gains the file's identity (device.inode, fileid_unix.go;
  empty elsewhere): two logs that begin alike are two files - the
  review-2026-08-followups row on collisions. A fingerprint saved before
  (hash alone) still matches by its hash (sameLog), so the first start after
  the upgrade resumes where it was.
- store.LogtailState reads back updated_at.

Tests: stop, append, rotate, append, start - both halves read, once each,
the earlier line not again; a mail.log.1 that is another file is not read
from the place; sameLog with an old and a new fingerprint.

CHANGELOG [Unreleased], audit plan line ticked, roadmap 23/62.
Evidence: GOOS=linux go vet ./internal/logtail clean; go test ./internal/logtail
ok on Windows but for TestFollowTailsAndRotates; the server run follows
before the push.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 16:25:29 +03:00
13 changed files with 422 additions and 54 deletions
+18
View File
@@ -5,6 +5,24 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version
## [Unreleased]
### Fixed
- **A log rotated while the panel was down is read to its end.** On start,
a saved place in a file that had been rotated meant "start on the new
file", and what Postfix wrote between the panel's stop and the rotation -
deliveries, refusals, deferrals - was never read. The tailer now reads the
rotated file (`mail.log.1`) from the saved place to its end first; when
that file is not the one the place belongs to (rotated twice), it says so
in the log, with the time from which nothing was read. The log's
fingerprint now includes the file's inode, so two files that begin alike
are still told apart (audit 2026-10-12, stage 2b).
- **A second deferral of the same second is counted after a restart.** The
Sender check's count of deferred mail remembered the time of the last
line it counted, to pass over lines read again after an unclean stop;
`mail.log` stamps whole seconds, so a second warning of that second was
passed over too. It now remembers the place in the log, saved with the
count in one transaction (audit 2026-10-12, stage 2b).
## [2.5.0] - 2026-10-14
Stages 1 and 2 of the 2026-10-12 audit: mail the delivery log showed wrong,
+2 -2
View File
@@ -736,7 +736,7 @@ stage follows as a cut of its own, with its own reviews. Five of the six
change one thing, the tailer and the state it saves, and are done in this
order - each stands on the one before.
- [ ] **A log rotated while the panel was down is read to its end** (owner,
- [x] **A log rotated while the panel was down is read to its end** (owner,
2026-10-13; the Not done list of from-header-binding had it as "deferrals
are lost", and it is wider). At start, a saved offset whose fingerprint
is not the current file's means "start on the new file"
@@ -751,7 +751,7 @@ order - each stands on the one before.
rotated twice: logged, with from when to when nothing was read. A test
that stops the tailer, appends, rotates, appends, starts it, and finds
both halves once. - **Opus**
- [ ] **The deferred counter remembers a place in the log, not a second**
- [x] **The deferred counter remembers a place in the log, not a second**
(owner, 2026-10-14, from the Not done list of from-header-binding; he
chose it over leaving the case written down). After an unclean stop the
tailer re-reads lines it had read, and the counter passes over every
+1 -1
View File
@@ -42,7 +42,7 @@ in `git log` and [CHANGELOG.md](../CHANGELOG.md).
| from-header-binding | Hold a message's `From:` header against the application that sends it | shipped `2.4.0` | 9/9 | [plans/from-header-binding.md](plans/from-header-binding.md) |
| panel-redesign | Panel redesign (accepted mockups, design contract) | shipped `2.0.0` | 27/27 | [plans/panel-redesign.md](plans/panel-redesign.md) |
| schema-squash | Squash SQLite migrations into a 2.x baseline | built in panel-redesign (stage 1) | — | [plans/panel-redesign.md](plans/panel-redesign.md) |
| audit-2026-10-12 | Follow-ups of the 2026-10-12 full-tree code audit, seven stages | in progress (stage 0 shipped `2.4.2`, stages 1+2 `2.5.0`; next stage 2b → `2.5.1`) | 22/62 | [audit-2026-10-12.md](audit-2026-10-12.md) |
| audit-2026-10-12 | Follow-ups of the 2026-10-12 full-tree code audit, seven stages | in progress (stage 0 shipped `2.4.2`, stages 1+2 `2.5.0`; next stage 2b → `2.5.1`) | 24/62 | [audit-2026-10-12.md](audit-2026-10-12.md) |
**Recommended order** (not binding; owner, 2026-10-09: feature order follows
development convenience): **panel-redesign** shipped as `2.0.0` (2026-10-11) and
+9
View File
@@ -0,0 +1,9 @@
//go:build !unix
package logtail
import "os"
// fileID is empty where the system has no inode to give (the panel runs on
// Linux; this keeps the package building elsewhere for its tests).
func fileID(*os.File) string { return "" }
+24
View File
@@ -0,0 +1,24 @@
//go:build unix
package logtail
import (
"fmt"
"os"
"syscall"
)
// fileID names the file itself - device and inode - which a rename keeps and
// a new file does not share. It is part of the log's fingerprint, so two log
// files whose first bytes happen to match are still told apart.
func fileID(f *os.File) string {
info, err := f.Stat()
if err != nil {
return ""
}
st, ok := info.Sys().(*syscall.Stat_t)
if !ok {
return ""
}
return fmt.Sprintf("%d.%d", st.Dev, st.Ino)
}
+17 -3
View File
@@ -162,7 +162,8 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
}
keeper := newLineKeeper(st)
return follow(ctx, path, &tracker{st: st, path: path}, caughtUp, keeper.flush, func(line string) {
tr := &tracker{st: st, path: path}
return follow(ctx, path, tr, caughtUp, keeper.flush, func(line string) {
// Until the first read reaches end-of-file the lines are those written
// while the panel was down - and, after an unclean stop, some it had
// read already.
@@ -177,7 +178,7 @@ func Run(ctx context.Context, path string, st StatusStore, retention RetentionDa
journal.handle(line, now, catchingUp)
}
if deferred != nil {
deferred.handle(line, now, catchingUp)
deferred.handle(line, now, catchingUp, tr.line)
}
keeper.add(line, now, catchingUp)
queueID, recipient, status, ok := parseDelivery(line)
@@ -569,6 +570,9 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
r *bufio.Reader
info os.FileInfo
pending string
// consumed is the offset just past what the reader has handed over,
// pending included: where a complete line ends.
consumed int64
)
openAt := func(offset int64, whence int) error {
nf, err := os.Open(path)
@@ -588,6 +592,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
f.Close()
}
f, r, info, pending = nf, bufio.NewReader(nf), ni, ""
consumed = offset
return nil
}
@@ -602,10 +607,17 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
case <-time.After(pollInterval):
}
}
if _, err := f.Seek(tr.resume(f), io.SeekStart); err != nil {
start := tr.resume(f)
if _, err := f.Seek(start, io.SeekStart); err != nil {
log.Printf("log-tailer: seek %s: %v", path, err)
start = 0
}
consumed = start
r.Reset(f) // the reader buffered from the pre-seek position
// Rotated while the panel was down: the end of the old file first.
if tr.readRotated(handle) {
flush()
}
defer func() {
if f != nil {
f.Close()
@@ -616,6 +628,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
defer flush()
for {
line, err := r.ReadString('\n')
consumed += int64(len(line))
if err == io.EOF {
pending += line // hold the partial line until it completes
return
@@ -626,6 +639,7 @@ func follow(ctx context.Context, path string, tr *tracker, caughtUp chan struct{
}
full := pending + line
pending = ""
tr.line = Place{FP: tr.fp, End: consumed}
handle(strings.TrimRight(full, "\r\n"))
}
}
+76
View File
@@ -815,3 +815,79 @@ func TestSplitTimestamp(t *testing.T) {
})
}
}
// A log rotated while the panel was down is read to its end before the new
// one: what Postfix wrote between the stop and the rotation is in mail.log.1
// (audit 2026-10-12, stage 2b). Both halves are read, once each.
func TestALogRotatedWhileDownIsReadToItsEnd(t *testing.T) {
old := pollInterval
pollInterval = 10 * time.Millisecond
t.Cleanup(func() { pollInterval = old })
dir := t.TempDir()
path := filepath.Join(dir, "mail.log")
seed := strings.Repeat("host postfix/qmgr[1]: seed line, not a delivery\n", 20)
if err := os.WriteFile(path, []byte(seed), 0o644); err != nil {
t.Fatal(err)
}
cs := &captureStore{}
stop := startRun(t, path, cs)
appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
stop()
// Down: Postfix writes, then logrotate renames and creates, and Postfix
// goes on in the new file.
appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, relay=x, dsn=2.0.0, status=sent (ok)")
if err := os.Rename(path, path+".1"); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, []byte(strings.Repeat("host postfix/qmgr[1]: the new log\n", 20)+
"host postfix/smtp[1]: Q3: to=<c@example.net>, relay=x, dsn=5.1.1, status=bounced (nope)\n"), 0o644); err != nil {
t.Fatal(err)
}
cs.reset()
stop = startRun(t, path, cs)
waitFor(t, func() bool {
got := cs.snapshot()
return contains(got, "Q2|b@example.net|sent") && contains(got, "Q3|c@example.net|bounced")
})
stop()
n := 0
for _, c := range cs.snapshot() {
if strings.HasPrefix(c, "Q1|") {
t.Errorf("Q1 was read again: %s", c)
}
if strings.HasPrefix(c, "Q2|") {
n++
}
}
if n != 1 {
t.Errorf("Q2 read %d times, want once", n)
}
}
// A mail.log.1 that is not the file the saved place belongs to is not read
// from that place: the new file is read from its start and the gap is said.
func TestARotatedLogThatIsNotTheSavedOneIsNotRead(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "mail.log")
if err := os.WriteFile(path+".1", []byte(strings.Repeat("host postfix/qmgr[1]: another file altogether\n", 20)+
"host postfix/smtp[1]: QX: to=<x@example.net>, relay=x, dsn=2.0.0, status=sent (ok)\n"), 0o644); err != nil {
t.Fatal(err)
}
tr := &tracker{path: path, behind: true, prev: store.LogtailState{Fingerprint: "not-this-one", Offset: 10}}
var read []string
if tr.readRotated(func(l string) { read = append(read, l) }) || len(read) != 0 {
t.Fatalf("read %d lines of a rotated file that is not the saved one", len(read))
}
}
// A fingerprint saved before it carried the file's identity still matches the
// file by its hash.
func TestSameLogAcceptsAFingerprintOfBefore(t *testing.T) {
if !sameLog("abc", "abc:1.2") || !sameLog("abc:1.2", "abc:1.2") || sameLog("abc:1.2", "abc:1.3") || sameLog("", "abc") {
t.Fatal("sameLog")
}
}
+115 -5
View File
@@ -1,12 +1,15 @@
package logtail
import (
"bufio"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
"log"
"os"
"strings"
"time"
"github.com/mixeme/selfpost/internal/store"
@@ -38,6 +41,39 @@ type tracker struct {
fp string // fingerprint of the file currently open ("" if too short)
saved int64 // last offset written to the store
lastSave time.Time
// behind is set by resume when the saved place belongs to a file that is
// no longer the log - rotated while the panel was down; prev is that place.
behind bool
prev store.LogtailState
// line is the place of the line being handled: set by follow and
// readRotated before each call of handle.
line Place
}
// Place is where a line of the log ends: the fingerprint of its file and the
// byte offset just past it. A rotated file keeps its fingerprint (the inode
// is part of it), so a place stays true across a rotation.
type Place struct {
FP string
End int64
}
// String is the place as it is saved: "<fingerprint>@<offset>".
func (p Place) String() string { return fmt.Sprintf("%s@%d", p.FP, p.End) }
// parsePlace reads a saved place; the zero Place for anything else.
func parsePlace(s string) Place {
i := strings.LastIndexByte(s, '@')
if i < 0 {
return Place{}
}
var end int64
if _, err := fmt.Sscan(s[i+1:], &end); err != nil {
return Place{}
}
return Place{FP: s[:i], End: end}
}
// resume returns the byte offset the tailer should start reading f at, having
@@ -68,7 +104,7 @@ func (t *tracker) resume(f *os.File) int64 {
case !ok:
t.saved = size
return size
case prev.Fingerprint != "" && prev.Fingerprint == t.fp && prev.Offset <= size:
case prev.Fingerprint != "" && sameLog(prev.Fingerprint, t.fp) && prev.Offset <= size:
t.saved = prev.Offset
if prev.Offset < size {
log.Printf("log-tailer: resuming %s at offset %d (%d bytes to catch up)",
@@ -78,10 +114,78 @@ func (t *tracker) resume(f *os.File) int64 {
default:
log.Printf("log-tailer: %s changed while the panel was down; reading from the start", t.path)
t.saved = 0
t.behind, t.prev = prev.Fingerprint != "", prev
return 0
}
}
// readRotated reads, when the log was rotated while the panel was down, the
// rest of the file the saved place belongs to: logrotate renamed it to
// mail.log.1 (delaycompress leaves it plain), and what Postfix wrote there
// between the panel's stop and the rotation - deliveries, refusals, deferrals -
// was otherwise never read (audit 2026-10-12, stage 2b). It is read from the
// saved offset to its end, line by line through handle, before the new file
// is read from its start. A mail.log.1 that is not that file - the log
// rotated more than once, or the file is gone - is said in the log, with the
// time from which nothing was read. It reports whether it read anything.
func (t *tracker) readRotated(handle func(string)) bool {
if !t.behind {
return false
}
t.behind = false
rotated := t.path + ".1"
lost := func(why string) {
log.Printf("log-tailer: %s; what was written to %s from %s until it was rotated was not read",
why, t.path, t.prev.UpdatedAt.UTC().Format(time.RFC3339))
}
f, err := os.Open(rotated)
if err != nil {
lost(fmt.Sprintf("%s cannot be opened (%v)", rotated, err))
return false
}
defer f.Close()
size, err := fileSize(f)
if err != nil || !sameLog(t.prev.Fingerprint, fingerprintOf(f)) || t.prev.Offset > size {
lost(fmt.Sprintf("%s is not the file the saved place belongs to (rotated more than once while the panel was down?)", rotated))
return false
}
if _, err := f.Seek(t.prev.Offset, io.SeekStart); err != nil {
lost(fmt.Sprintf("seek %s: %v", rotated, err))
return false
}
fp := fingerprintOf(f)
end := t.prev.Offset
r := bufio.NewReader(f)
for {
line, err := r.ReadString('\n')
end += int64(len(line))
if line != "" {
t.line = Place{FP: fp, End: end}
handle(strings.TrimRight(line, "\r\n"))
}
if err != nil {
break
}
}
log.Printf("log-tailer: read %s from offset %d to its end (%d bytes) before the new %s",
rotated, t.prev.Offset, size-t.prev.Offset, t.path)
return true
}
// sameLog says whether a saved fingerprint is the current file's. One saved
// before the fingerprint carried the file's identity is its hash alone, and is
// compared with the hash.
func sameLog(saved, current string) bool {
if saved == current {
return true
}
if !strings.Contains(saved, ":") {
hash, _, _ := strings.Cut(current, ":")
return saved != "" && saved == hash
}
return false
}
// adopt re-fingerprints after the follow loop switched to a rotated-in file and
// persists the fresh start immediately, so a restart right after a rotation
// does not resume at the old file's offset.
@@ -113,9 +217,11 @@ func (t *tracker) record(f *os.File, offset int64, force bool) {
t.lastSave = time.Now()
}
// fingerprintOf hashes the head of the file. It returns "" for a file too short
// to identify — the head would still change as Postfix appends, so such a
// fingerprint could not be compared meaningfully on the next start.
// fingerprintOf hashes the head of the file and adds the file's identity
// (device and inode, where there is one): two log files whose first bytes
// match are still two files. It returns "" for a file too short to identify —
// the head would still change as Postfix appends, so such a fingerprint could
// not be compared meaningfully on the next start.
func fingerprintOf(f *os.File) string {
buf := make([]byte, fingerprintSize)
n, err := f.ReadAt(buf, 0)
@@ -127,7 +233,11 @@ func fingerprintOf(f *os.File) string {
return ""
}
sum := sha256.Sum256(buf)
return hex.EncodeToString(sum[:])
fp := hex.EncodeToString(sum[:])
if id := fileID(f); id != "" {
fp += ":" + id
}
return fp
}
func fileSize(f *os.File) (int64, error) {
+40 -17
View File
@@ -46,8 +46,11 @@ import (
// SenderCheckStore is where the count goes, and where the time of the last
// counted line is kept. *store.Store satisfies it.
type SenderCheckStore interface {
AddSenderCheckDeferred(at time.Time, n int64) error
// AddSenderCheckDeferred counts and saves the place in the log reached,
// in one transaction.
AddSenderCheckDeferred(at time.Time, n int64, place string) error
SenderCheckDeferredAt() (time.Time, error)
SenderCheckCountedPlace() (string, error)
}
// SenderCheck turns the counting on: the store and the socket Postfix was
@@ -72,13 +75,16 @@ type deferredCounter struct {
// listening, and a reply that did not come.
connect, broken string
failed map[string]bool // smtpd pid -> its current client was counted already
// last is the time of the latest line counted, by this run or an earlier
// one. While the tailer catches up after a start it may be reading lines
// it had read before an unclean stop; a warning no later than this was
// counted then. mail.log's stamps are whole seconds, so a warning of the
// very second of the last one counted can be passed over - once, at a
// restart - which is the cheaper mistake than counting a stretch twice.
last time.Time
// counted is the place in the log of the latest warning counted, by this
// run or an earlier one, saved with the count. While the tailer catches up
// after a start it may be reading lines it had read before an unclean
// stop; a warning at or before that place was counted then. Until 2.5.1
// this was a time, and mail.log's stamps are whole seconds: a second
// warning of the same second was passed over (audit 2026-10-12, stage
// 2b). A place in another file - read again from a rotated log - has no
// order to compare with, and last, its time, decides.
counted Place
last time.Time
}
func newDeferredCounter(sc *SenderCheck) *deferredCounter {
@@ -86,8 +92,13 @@ func newDeferredCounter(sc *SenderCheck) *deferredCounter {
if err != nil {
log.Printf("log-tailer: sender check statistics: %v", err)
}
place, err := sc.Store.SenderCheckCountedPlace()
if err != nil {
log.Printf("log-tailer: sender check statistics: %v", err)
}
return &deferredCounter{
last: last,
counted: parsePlace(place),
st: sc.Store,
connect: "warning: connect to Milter service unix:" + sc.Socket + ":",
broken: "warning: milter unix:" + sc.Socket + ":",
@@ -95,9 +106,9 @@ func newDeferredCounter(sc *SenderCheck) *deferredCounter {
}
}
// handle looks at one mail.log line. catchingUp is true until the tailer has
// read to the end of the log for the first time.
func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) {
// handle looks at one mail.log line, which ends at place. catchingUp is true
// until the tailer has read to the end of the log for the first time.
func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool, place Place) {
at, rest := lineTime(line, now)
m := milterLineRe.FindStringSubmatch(rest)
if m == nil {
@@ -107,7 +118,7 @@ func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) {
if program == "cleanup" {
// A message cleanup could not finish because the check did not reply.
if strings.HasPrefix(text, c.broken) {
c.count(at, catchingUp)
c.count(at, catchingUp, place)
}
return
}
@@ -126,19 +137,31 @@ func (c *deferredCounter) handle(line string, now time.Time, catchingUp bool) {
c.failed = make(map[string]bool)
}
c.failed[pid] = true
c.count(at, catchingUp)
c.count(at, catchingUp, place)
}
}
// count adds one deferred message at at, unless it is a line read again.
func (c *deferredCounter) count(at time.Time, catchingUp bool) {
if catchingUp && !at.After(c.last) {
// count adds one deferred message at at, the line ending at place, unless it
// is a line read again.
func (c *deferredCounter) count(at time.Time, catchingUp bool, place Place) {
if catchingUp && c.countedAlready(at, place) {
return
}
if at.After(c.last) {
c.last = at
}
if err := c.st.AddSenderCheckDeferred(at, 1); err != nil {
c.counted = place
if err := c.st.AddSenderCheckDeferred(at, 1, place.String()); err != nil {
log.Printf("log-tailer: sender check statistics: %v", err)
}
}
// countedAlready says whether a warning read while catching up was counted
// before: by its place when it is in the file of the last one counted, by its
// time otherwise.
func (c *deferredCounter) countedAlready(at time.Time, place Place) bool {
if place.FP != "" && place.FP == c.counted.FP {
return place.End <= c.counted.End
}
return !at.After(c.last)
}
+50 -6
View File
@@ -6,13 +6,17 @@ import (
)
type deferredStore struct {
days map[string]int64
last time.Time
days map[string]int64
last time.Time
place string
}
func (d *deferredStore) SenderCheckDeferredAt() (time.Time, error) { return d.last, nil }
func (d *deferredStore) AddSenderCheckDeferred(at time.Time, n int64) error {
func (d *deferredStore) SenderCheckCountedPlace() (string, error) { return d.place, nil }
func (d *deferredStore) AddSenderCheckDeferred(at time.Time, n int64, place string) error {
d.place = place
if d.days == nil {
d.days = make(map[string]int64)
}
@@ -77,7 +81,7 @@ func TestDeferredCounter(t *testing.T) {
// cleanup that writes it.
`2026-10-13T00:01:00.000000+00:00 mail postfix/cleanup[1195]: B6B484831946: 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=<x>`,
} {
c.handle(line, now, false)
c.handle(line, now, false, Place{})
}
if len(st.days) != 2 || st.days["2026-10-12"] != 1 || st.days["2026-10-13"] != 2 {
t.Fatalf("deferred per day = %v, want one on the first day and two on the second", st.days)
@@ -100,7 +104,7 @@ func TestDeferredCounterDoesNotCountALogReadAgain(t *testing.T) {
c := newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"})
for _, line := range first {
c.handle(line, now, false)
c.handle(line, now, false, Place{})
}
if st.days["2026-10-12"] != 2 {
t.Fatalf("first run counted %d, want 2", st.days["2026-10-12"])
@@ -109,9 +113,49 @@ func TestDeferredCounterDoesNotCountALogReadAgain(t *testing.T) {
// A new run: the same lines again, then what the outage added.
c = newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"})
for _, line := range append(first, outage("13", "2026-10-12T16:00:09.00000")...) {
c.handle(line, now, true)
c.handle(line, now, true, Place{})
}
if st.days["2026-10-12"] != 3 {
t.Fatalf("after the restart the day has %d, want 3", st.days["2026-10-12"])
}
}
// Two warnings of the same second, the panel stopped uncleanly between them:
// the next start reads both again, and the place of the first - not its
// second - tells them apart. Each is counted once (audit 2026-10-12, stage
// 2b; a time alone passed the second over).
func TestDeferredCounterRemembersAPlaceNotASecond(t *testing.T) {
st := &deferredStore{}
now := time.Date(2026, 10, 12, 18, 0, 0, 0, time.UTC)
lines := []string{
`2026-10-12T16:00:00+00:00 mail postfix/smtpd[11]: connect from unknown[10.77.0.1]`,
`2026-10-12T16:00:01+00:00 mail postfix/smtpd[11]: warning: connect to Milter service unix:/run/selfpost/sender.sock: Connection refused`,
`2026-10-12T16:00:01+00:00 mail postfix/smtpd[12]: connect from unknown[10.77.0.1]`,
`2026-10-12T16:00:01+00:00 mail postfix/smtpd[12]: warning: connect to Milter service unix:/run/selfpost/sender.sock: Connection refused`,
}
// The place each line ends at, in one file.
places := make([]Place, len(lines))
end := int64(0)
for i, l := range lines {
end += int64(len(l)) + 1
places[i] = Place{FP: "abc:1.2", End: end}
}
c := newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"})
for i := 0; i < 2; i++ { // stopped after the first warning
c.handle(lines[i], now, false, places[i])
}
if st.days["2026-10-12"] != 1 {
t.Fatalf("before the stop: %d, want 1", st.days["2026-10-12"])
}
// Started again: the saved offset was from before the stop, so all four
// lines are read again.
c = newDeferredCounter(&SenderCheck{Store: st, Socket: "/run/selfpost/sender.sock"})
for i, l := range lines {
c.handle(l, now, true, places[i])
}
if st.days["2026-10-12"] != 2 {
t.Fatalf("after the restart the day has %d, want 2", st.days["2026-10-12"])
}
}
+9 -3
View File
@@ -14,17 +14,23 @@ import (
type LogtailState struct {
Fingerprint string
Offset int64
// UpdatedAt is when the position was saved; read back only (zero on save).
UpdatedAt time.Time
}
// 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
var (
st LogtailState
updated string
)
err := s.db.QueryRow(
`SELECT fingerprint, read_offset FROM logtail_state WHERE path = ?`,
`SELECT fingerprint, read_offset, updated_at FROM logtail_state WHERE path = ?`,
path,
).Scan(&st.Fingerprint, &st.Offset)
).Scan(&st.Fingerprint, &st.Offset, &updated)
st.UpdatedAt, _ = time.Parse(time.RFC3339, updated)
if errors.Is(err, sql.ErrNoRows) {
return LogtailState{}, false, nil
}
+39 -14
View File
@@ -30,6 +30,9 @@ const (
// settingSenderCheckDeferredAt is when a message was last deferred because
// the check did not answer (RFC3339 UTC).
settingSenderCheckDeferredAt = "sender_check_deferred_at"
// settingSenderCheckCountedPlace is the place in mail.log of the last
// warning counted ("<fingerprint>@<offset>"), saved with the count.
settingSenderCheckCountedPlace = "sender_check_counted_place"
)
// senderCheckKeepDays is how long a day of the statistics is kept.
@@ -80,33 +83,55 @@ func (s *Store) AddSenderCheckDown(at time.Time, d time.Duration) error {
}
// AddSenderCheckDeferred counts, on the day of at, n messages deferred because
// the check did not answer, and remembers at as the last such moment.
func (s *Store) AddSenderCheckDeferred(at time.Time, n int64) error {
// the check did not answer, remembers at as the last such moment, and saves
// place - where in mail.log the counted line ends - in the same transaction:
// a count without its place, or the other way round, would be counted again
// or lost at the next start (audit 2026-10-12, stage 2b).
func (s *Store) AddSenderCheckDeferred(at time.Time, n int64, place string) error {
if n <= 0 {
return nil
}
_, err := s.db.Exec(
`INSERT INTO sender_check_days (day, deferred) VALUES (?, ?)
ON CONFLICT (day) DO UPDATE SET deferred = deferred + excluded.deferred`,
at.UTC().Format(dayLayout), n,
)
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("sender check deferred: %w", err)
}
// Lines are read in order, but a re-read log must not move the mark back.
last, err := s.GetSetting(settingSenderCheckDeferredAt)
if err != nil {
return err
defer tx.Rollback()
if _, err := tx.Exec(
`INSERT INTO sender_check_days (day, deferred) VALUES (?, ?)
ON CONFLICT (day) DO UPDATE SET deferred = deferred + excluded.deferred`,
at.UTC().Format(dayLayout), n,
); err != nil {
return fmt.Errorf("sender check deferred: %w", err)
}
// To the fraction of a second the log gives, and of a fixed width so that
// the strings compare as the times do.
// the strings compare as the times do; a re-read log must not move the
// mark back.
stamp := at.UTC().Format("2006-01-02T15:04:05.000000Z")
if stamp > last {
return s.SetSetting(settingSenderCheckDeferredAt, stamp)
setting := func(key, value, cond string) error {
_, err := tx.Exec(`INSERT INTO settings (key, value) VALUES (?, ?)
ON CONFLICT (key) DO UPDATE SET value = excluded.value`+cond, key, value)
return err
}
if err := setting(settingSenderCheckDeferredAt, stamp, " WHERE excluded.value > settings.value"); err != nil {
return fmt.Errorf("sender check deferred: %w", err)
}
if place != "" {
if err := setting(settingSenderCheckCountedPlace, place, ""); err != nil {
return fmt.Errorf("sender check deferred: %w", err)
}
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("sender check deferred: %w", err)
}
return nil
}
// SenderCheckCountedPlace is the place in mail.log of the last warning
// counted; "" before any was.
func (s *Store) SenderCheckCountedPlace() (string, error) {
return s.GetSetting(settingSenderCheckCountedPlace)
}
// SenderCheckDeferredAt is when a message was last counted as deferred; zero
// if none was.
func (s *Store) SenderCheckDeferredAt() (time.Time, error) {
+22 -3
View File
@@ -42,12 +42,12 @@ func TestSenderCheckStats(t *testing.T) {
must(st.AddSenderCheckDown(now, 3*time.Second))
must(st.AddSenderCheckDown(now, 1500*time.Millisecond)) // whole seconds: one
must(st.AddSenderCheckDown(now, 400*time.Millisecond)) // nothing
must(st.AddSenderCheckDeferred(now.Add(-time.Hour), 2))
must(st.AddSenderCheckDeferred(now.Add(-time.Hour), 2, ""))
must(st.AddSenderCheckDown(day(10), 30*time.Second))
must(st.AddSenderCheckDeferred(day(10), 1))
must(st.AddSenderCheckDeferred(day(10), 1, ""))
must(st.AddSenderCheckDown(day(29), time.Minute))
must(st.AddSenderCheckDown(day(30), time.Hour)) // the 31st day back: the year's only
must(st.AddSenderCheckDeferred(day(200), 5))
must(st.AddSenderCheckDeferred(day(200), 5, ""))
must(st.AddSenderCheckDown(day(365), 24*time.Hour)) // a year and a day: nobody's
got, err = st.SenderCheckStats(now)
@@ -112,3 +112,22 @@ func TestRefusedRowKeepsItsReason(t *testing.T) {
t.Errorf("GetSendLog = %+v, %v", one, err)
}
}
// The place in the log is saved with the count, and the moment of the last
// deferral does not move back when a log is read again.
func TestSenderCheckDeferredSavesItsPlace(t *testing.T) {
st := openTestStore(t)
later := time.Date(2026, 10, 14, 12, 0, 1, 0, time.UTC)
if err := st.AddSenderCheckDeferred(later, 1, "abc:1.2@100"); err != nil {
t.Fatal(err)
}
if err := st.AddSenderCheckDeferred(later.Add(-time.Hour), 1, "abc:1.2@200"); err != nil {
t.Fatal(err)
}
if p, err := st.SenderCheckCountedPlace(); err != nil || p != "abc:1.2@200" {
t.Fatalf("place = %q, %v", p, err)
}
if at, err := st.SenderCheckDeferredAt(); err != nil || !at.Equal(later) {
t.Fatalf("last deferral = %s, %v; want %s", at, err, later)
}
}