Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c452606175 | ||
|
|
4678015bbd |
@@ -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,
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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 "" }
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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"))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user