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>
This commit is contained in:
mix
2026-10-14 16:28:13 +03:00
co-authored by Claude Opus 5.5
parent 4678015bbd
commit c452606175
9 changed files with 204 additions and 45 deletions
+13 -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,9 +607,12 @@ 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) {
@@ -620,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
@@ -630,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"))
}
}
+32
View File
@@ -46,6 +46,34 @@ type tracker struct {
// 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
@@ -125,10 +153,14 @@ func (t *tracker) readRotated(handle func(string)) bool {
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 {
+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"])
}
}