diff --git a/CHANGES.md b/CHANGES.md index 527f9f1..93feab7 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,4 +1,14 @@ # Unreleased +* PR #6: fix stale watcher state across log rotation. A watch armed + while the file was being replaced took its size baseline from the old + file, so the first write to the new file looked like a truncation and + the spurious reopen duplicated already-delivered lines. A watch kept + across the pendingReopen/Truncated reopen paths could be dead with a + latched Deleted notification, causing a bogus second reopen (line + duplication) — reopen paths now drop and re-arm the watch. A watch + removed externally (Cleanup) made the watcher goroutine exit silently + and corrupted the shared refcount, hanging the tailer and every later + tail of the same name. # Version v1.4.14 * PR #4: re-check the file right after arming the inotify watch. The diff --git a/go.mod b/go.mod index 7780c64..b6bd04f 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,13 @@ go 1.25.7 require ( github.com/fsnotify/fsnotify v1.6.0 + github.com/stretchr/testify v1.11.1 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 ) -require golang.org/x/sys v0.0.0-20220908164124-27713097b956 // indirect +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + golang.org/x/sys v0.0.0-20220908164124-27713097b956 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum index 32326e9..3cd1493 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,16 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= golang.org/x/sys v0.0.0-20220908164124-27713097b956 h1:XeJjHH1KiLpKGb6lvMiksZ9l0fVUh+AmGcm0nOMEBOY= golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/tail.go b/tail.go index 27d00ef..b651809 100644 --- a/tail.go +++ b/tail.go @@ -128,6 +128,11 @@ var ( // stays zero outside of tests. var delayBeforeWatch time.Duration +// delayBeforeRecheck stalls a tailer between arming the watch and the +// recheck that follows, so that tests can reproduce a rotation landing +// in that window. Like delayBeforeWatch, it stays zero outside of tests. +var delayBeforeRecheck time.Duration + // TailFile begins tailing the file. And returns a pointer to a Tail struct // and an error. An output stream is made available via the Tail.Lines // channel (e.g. to be looped and printed). To handle errors during tailing, @@ -381,6 +386,18 @@ func (tail *Tail) tailFileSync() { } } +// dropWatch tears down the current watch subscription, if any, so that +// the next waitForChanges arms a fresh one and rechecks the file. The +// producer goroutine is stopped synchronously before the re-arm: an +// abandoned live producer would keep consuming from the shared per-file +// events channel, stealing events from the new subscription. +func (tail *Tail) dropWatch() { + if tail.changes != nil { + tail.changes.Stop() + tail.changes = nil + } +} + // waitForChanges waits until the file has been appended, deleted, // moved or truncated. Truncated files are always reopened. // @@ -393,7 +410,11 @@ func (tail *Tail) tailFileSync() { func (tail *Tail) waitForChanges() error { if tail.pendingDelete { tail.pendingDelete = false - tail.changes = nil + // The reopen below resolves a pending replacement as well; a + // stale pendingReopen left set would trigger a second reopen + // and re-deliver everything read since this one. + tail.pendingReopen = false + tail.dropWatch() if tail.ReOpen { // XXX: we must not log from a library. tail.Logger.Printf("Re-opening moved/deleted file %s ...", tail.Filename) @@ -410,8 +431,14 @@ func (tail *Tail) waitForChanges() error { if tail.pendingReopen { tail.pendingReopen = false - // The watch is already on the file that occupies the name now, so - // only the descriptor has to catch up with it. Keep tail.changes. + // The watch cannot be trusted across the switch: it may sit on + // the replacing file with a stale size baseline, or the rename + // may have already killed the producer, leaving behind a + // latched Deleted notification that would trigger a bogus + // second reopen. Drop the subscription and re-arm it after the + // reopen; the recheck that follows the re-arm picks up + // anything that happened in between. + tail.dropWatch() tail.Logger.Printf("Re-opening replaced file %s ...", tail.Filename) if err := tail.reopen(); err != nil { return err @@ -439,6 +466,10 @@ func (tail *Tail) waitForChanges() error { return err } + if delayBeforeRecheck > 0 { + time.Sleep(delayBeforeRecheck) + } + recheck, err := tail.recheckAfterWatch(pos) if err != nil { return err @@ -463,6 +494,9 @@ func (tail *Tail) waitForChanges() error { return nil case <-tail.changes.Truncated: // Always reopen truncated files (Follow is true) + // The descriptor moves to whatever occupies the name now, so + // the watch has to be dropped and re-armed with it. + tail.dropWatch() tail.Logger.Printf("Re-opening truncated file %s ...", tail.Filename) if err := tail.reopen(); err != nil { return err @@ -509,6 +543,9 @@ func (tail *Tail) recheckAfterWatch(pos int64) (bool, error) { return true, nil } + // pendingReopen reports "read now" on purpose: the extra read cycle + // drains what is left in the open descriptor before waitForChanges + // switches it to the file that took over the name. return tail.pendingReopen, nil } diff --git a/watch/filechanges.go b/watch/filechanges.go index 9e6b8d4..68bd324 100644 --- a/watch/filechanges.go +++ b/watch/filechanges.go @@ -2,15 +2,43 @@ // Copyright (c) 2019 FOSS contributors of https://github.com/nxadm/tail package watch +import "sync" + type FileChanges struct { Modified chan bool // Channel to get notified of modifications Truncated chan bool // Channel to get notified of truncations Deleted chan bool // Channel to get notified of deletions/renames + + // stop asks the producing watcher goroutine to quit; stopped is + // closed by the producer once it has. Together they let a consumer + // drop its subscription synchronously, so that a fresh watch can + // be armed without the old producer competing for the same shared + // events channel. + stop chan struct{} + stopped chan struct{} + stopOnce sync.Once + hasProducer bool } func NewFileChanges() *FileChanges { return &FileChanges{ - make(chan bool, 1), make(chan bool, 1), make(chan bool, 1)} + Modified: make(chan bool, 1), + Truncated: make(chan bool, 1), + Deleted: make(chan bool, 1), + stop: make(chan struct{}), + stopped: make(chan struct{}), + } +} + +// Stop asks the producing watcher goroutine to quit and waits until it +// has done so. It is safe to call Stop multiple times and after the +// producer has already quit on its own. A FileChanges that never had a +// producer attached stops right away. +func (fc *FileChanges) Stop() { + fc.stopOnce.Do(func() { close(fc.stop) }) + if fc.hasProducer { + <-fc.stopped + } } func (fc *FileChanges) NotifyModified() { diff --git a/watch/inotify.go b/watch/inotify.go index 274e631..49ecd3c 100644 --- a/watch/inotify.go +++ b/watch/inotify.go @@ -77,15 +77,25 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChange } changes := NewFileChanges() - fw.Size = pos + + // Seed the size baseline and the inode from the file that occupies + // the name right now, not from the caller's offset: after a rotation + // in the unwatched window the name already points at a different, + // usually smaller file, and a stale baseline would make the first + // write to it look like a truncation, causing a spurious reopen that + // re-delivers already-sent lines. Changes that land before the watch + // is armed are the caller's recheck's job to detect, not ours. + fw.Size = 0 + fw.inodeId = 0 var stat syscall.Stat_t - // Record file inodeId. - err = syscall.Stat(fw.Filename, &stat) - if err == nil { + if err := syscall.Stat(fw.Filename, &stat); err == nil { + fw.Size = stat.Size fw.inodeId = stat.Ino } + changes.hasProducer = true go func() { + defer close(changes.stopped) events := Events(fw.Filename) @@ -98,12 +108,21 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChange select { case evt, ok = <-events: if !ok { - RemoveWatch(fw.Filename) + // The events channel is closed by an external + // RemoveWatch (e.g. Cleanup): the watch is gone + // already and removing it again would corrupt + // the shared refcount. Report the file as gone + // so the consumer re-arms instead of blocking + // forever on a subscription that cannot fire. + changes.NotifyDeleted() return } case <-t.Dying(): RemoveWatch(fw.Filename) return + case <-changes.stop: + RemoveWatch(fw.Filename) + return } switch { @@ -148,7 +167,6 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChange } changes.NotifyTruncated() } - prevSize = fw.Size } } }() diff --git a/watch/inotify_tracker.go b/watch/inotify_tracker.go index 0ca90f5..448a3c3 100644 --- a/watch/inotify_tracker.go +++ b/watch/inotify_tracker.go @@ -177,10 +177,16 @@ func (shared *InotifyTracker) removeWatch(winfo *watchInfo) error { // Watch for new files to be created in the parent directory. fname = filepath.Dir(fname) } - shared.watchNums[fname]-- - watchNum := shared.watchNums[fname] - if watchNum == 0 { - delete(shared.watchNums, fname) + // A duplicate removal must not drive the refcount negative: a watch + // armed later would then skip the fsnotify subscription and never + // receive a single event. + lastRef := false + if shared.watchNums[fname] > 0 { + shared.watchNums[fname]-- + if shared.watchNums[fname] == 0 { + delete(shared.watchNums, fname) + lastRef = true + } } shared.mux.Unlock() @@ -189,7 +195,7 @@ func (shared *InotifyTracker) removeWatch(winfo *watchInfo) error { // This needs to happen after releasing the lock because fsnotify waits // synchronously for the kernel to acknowledge the removal of the watch // for this file, which causes us to deadlock if we still held the lock. - if watchNum == 0 { + if lastRef { err = shared.watcher.Remove(fname) } diff --git a/watch/polling.go b/watch/polling.go index 9506118..9e2ba86 100644 --- a/watch/polling.go +++ b/watch/polling.go @@ -56,18 +56,23 @@ func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChange // XXX: use tomb.Tomb to cleanly manage these goroutines. replace // the fatal (below) with tomb's Kill. - fw.Size = pos + // Seed the size baseline from the file itself, not from the caller's + // offset: see the same seeding in InotifyFileWatcher.ChangeEvents. + fw.Size = origFi.Size() + changes.hasProducer = true go func() { + defer close(changes.stopped) + prevSize := fw.Size for { select { + case <-time.After(POLL_DURATION): case <-t.Dying(): return - default: + case <-changes.stop: + return } - - time.Sleep(POLL_DURATION) fi, err := os.Stat(fw.Filename) if err != nil { // Windows cannot delete a file if a handle is still open (tail keeps one open) diff --git a/watch/watch.go b/watch/watch.go index 9609599..c632fb7 100644 --- a/watch/watch.go +++ b/watch/watch.go @@ -13,10 +13,13 @@ type FileWatcher interface { BlockUntilExists(*tomb.Tomb) error // ChangeEvents reports on changes to a file, be it modification, - // deletion, renames or truncations. Returned FileChanges group of - // channels will be closed, thus become unusable, after a deletion - // or truncation event. - // In order to properly report truncations, ChangeEvents requires - // the caller to pass their current offset in the file. + // deletion, renames or truncations. The watcher takes its size + // baseline from the file that occupies the name at arm time, so + // changes that happened before the watch was armed are the + // caller's job to detect. After a deletion event the producing + // goroutine quits and the FileChanges must be discarded; + // FileChanges.Stop releases the producer explicitly when the + // caller wants to re-arm the watch. + // The offset argument is unused and kept for compatibility. ChangeEvents(*tomb.Tomb, int64) (*FileChanges, error) } diff --git a/watch_baseline_test.go b/watch_baseline_test.go new file mode 100644 index 0000000..42c0bf5 --- /dev/null +++ b/watch_baseline_test.go @@ -0,0 +1,100 @@ +// Copyright (c) 2026 FOSS contributors of https://github.com/tarantool/go-tail + +package tail + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// dupSettle is how long the tests wait for a reopen to settle or for a +// spurious duplicate line to show up. +const dupSettle = time.Second + +// expectNoLine asserts that no line arrives on the tailer within d. +func expectNoLine(t *testing.T, tailer *Tail, d time.Duration) { + t.Helper() + + select { + case line, ok := <-tailer.Lines: + require.True(t, ok, "Lines channel closed unexpectedly") + require.Failf(t, "unexpected extra line", "got %q (err=%v)", line.Text, line.Err) + case <-time.After(d): + } +} + +// eachWatcher runs the test against both watcher implementations. +func eachWatcher(t *testing.T, test func(t *testing.T, poll bool)) { + t.Helper() + + for _, tc := range []struct { + name string + poll bool + }{ + {name: "inotify", poll: false}, + {name: "polling", poll: true}, + } { + t.Run(tc.name, func(t *testing.T) { + test(t, tc.poll) + }) + } +} + +// The file is replaced while the watch is being armed, so the watch ends +// up on the file that took over the name. A later append to that file +// must not be mistaken for a truncation: the watcher used to seed its +// size baseline with the tailer's offset in the old file, and the +// smaller new file looked truncated on its first write, triggering a +// reopen and a re-read from offset zero that duplicated every line +// delivered from the new file so far. +func TestReplaceWhileArmingThenAppend(t *testing.T) { + eachWatcher(t, func(t *testing.T, poll bool) { + stallBeforeWatch(t) + + dir := t.TempDir() + path := filepath.Join(dir, "rotated.log") + writeFile(t, path, "a rather long line that moves the offset far ahead\n") + + tailer := startTail(t, path, Config{Follow: true, ReOpen: true, Poll: poll}) + expectLine(t, tailer, "a rather long line that moves the offset far ahead") + + time.Sleep(windowSettle) + require.NoError(t, os.Rename(path, path+".bak")) + writeFile(t, path, "n1\n") + expectLine(t, tailer, "n1") + + // Let the reopen settle, then append: the append must produce + // exactly one new line and no replay of the lines above. + time.Sleep(dupSettle) + appendFile(t, path, "n2\n") + expectLine(t, tailer, "n2") + expectNoLine(t, tailer, dupSettle) + }) +} + +// Same stale-baseline scenario as above, but the file is truncated and +// rewritten in place instead of being replaced. +func TestTruncateWhileArmingThenAppend(t *testing.T) { + eachWatcher(t, func(t *testing.T, poll bool) { + stallBeforeWatch(t) + + path := filepath.Join(t.TempDir(), "truncated.log") + writeFile(t, path, "a rather long line that moves the offset far ahead\n") + + tailer := startTail(t, path, Config{Follow: true, ReOpen: true, Poll: poll}) + expectLine(t, tailer, "a rather long line that moves the offset far ahead") + + time.Sleep(windowSettle) + writeFile(t, path, "short\n") + expectLine(t, tailer, "short") + + time.Sleep(dupSettle) + appendFile(t, path, "next\n") + expectLine(t, tailer, "next") + expectNoLine(t, tailer, dupSettle) + }) +} diff --git a/watch_lifecycle_test.go b/watch_lifecycle_test.go new file mode 100644 index 0000000..3befd64 --- /dev/null +++ b/watch_lifecycle_test.go @@ -0,0 +1,174 @@ +// Copyright (c) 2026 FOSS contributors of https://github.com/tarantool/go-tail + +package tail + +import ( + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/tarantool/go-tail/watch" +) + +func stallBeforeRecheck(t *testing.T) { + t.Helper() + + delayBeforeRecheck = watchDelay + t.Cleanup(func() { delayBeforeRecheck = 0 }) +} + +// appendLine is an error-returning counterpart of appendFile for use +// outside of the test goroutine. +func appendLine(path, line string) error { + file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + defer file.Close() + + _, err = file.WriteString(line) + return err +} + +// The file is replaced after the watch was armed on the old file but +// before the tailer compared the descriptor with the name. By the time +// the tailer switches to the new file the watcher is already dead: the +// rename made it drop the watch and quit, leaving a latched Deleted +// notification behind. The tailer used to keep that dead subscription +// and honour the stale notification, reopening the already-reopened +// file a second time and re-delivering everything read in between. +func TestReplaceWhileRecheckingThenAppend(t *testing.T) { + eachWatcher(t, func(t *testing.T, poll bool) { + stallBeforeRecheck(t) + + dir := t.TempDir() + path := filepath.Join(dir, "rotated.log") + writeFile(t, path, "before rotation\n") + + tailer := startTail(t, path, Config{Follow: true, ReOpen: true, Poll: poll}) + expectLine(t, tailer, "before rotation") + + // Give the tailer time to arm the watch and enter the stall, + // then rotate inside the arm-to-recheck window. + time.Sleep(windowSettle) + require.NoError(t, os.Rename(path, path+".bak")) + writeFile(t, path, "n1\n") + expectLine(t, tailer, "n1") + expectNoLine(t, tailer, dupSettle) + + appendFile(t, path, "n2\n") + expectLine(t, tailer, "n2") + expectNoLine(t, tailer, dupSettle) + }) +} + +// Someone removes the watch out from under an armed tailer (Cleanup is +// documented as a process-exit helper, but nothing stops it from being +// called earlier). The watcher goroutine used to treat the closed events +// channel as a silent exit: it removed the already-removed watch again, +// corrupting the shared refcount, and told nobody, leaving the tailer +// blocked forever on a subscription that could not fire anymore. It must +// report the file as gone instead, so the tailer re-arms and keeps +// delivering lines. Only the inotify watcher shares state this way. +func TestExternalWatchRemovalDoesNotHang(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "cleaned.log") + writeFile(t, path, "seed\n") + + tailer := startTail(t, path, Config{Follow: true, ReOpen: true}) + expectLine(t, tailer, "seed") + + // Let the tailer arm the watch, then yank the watch away. + time.Sleep(windowSettle) + require.NoError(t, watch.Cleanup(path)) + + appendFile(t, path, "n1\n") + + // The recovery reopen legitimately re-reads the file from the + // start; the only requirement is that n1 arrives instead of the + // tailer hanging forever. + deadline := time.After(lineWait) + for { + select { + case line, ok := <-tailer.Lines: + require.True(t, ok, "Lines channel closed while waiting for %q", "n1") + require.NoError(t, line.Err) + if line.Text == "n1" { + return + } + case <-deadline: + t.Fatal("tailer hung after external watch removal") + } + } +} + +// Rotation stress: lines are appended in small batches with a rotation +// after every batch, racing the tailer's EOF/arm/recheck cycle. Every +// line must be delivered exactly once, in order. +func TestRotationStressExactlyOnce(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "app.log") + writeFile(t, path, "seed\n") + + tailer := startTail(t, path, Config{Follow: true, ReOpen: true}) + expectLine(t, tailer, "seed") + + const ( + rotations = 20 + batch = 5 + ) + + var expected []string + for r := 0; r < rotations; r++ { + for l := 0; l < batch; l++ { + expected = append(expected, fmt.Sprintf("r%02d-l%02d", r, l)) + } + } + expected = append(expected, "final") + + writeErr := make(chan error, 1) + go func() { + writeErr <- func() error { + for r := 0; r < rotations; r++ { + for l := 0; l < batch; l++ { + if err := appendLine(path, fmt.Sprintf("r%02d-l%02d\n", r, l)); err != nil { + return err + } + time.Sleep(5 * time.Millisecond) + } + if err := os.Rename(path, fmt.Sprintf("%s.%d", path, r)); err != nil { + return err + } + if err := os.WriteFile(path, nil, 0o644); err != nil { + return err + } + } + return appendLine(path, "final\n") + }() + }() + + var got []string + deadline := time.After(30 * time.Second) +loop: + for { + select { + case line, ok := <-tailer.Lines: + require.True(t, ok, "Lines channel closed mid-stress") + require.NoError(t, line.Err) + got = append(got, line.Text) + if line.Text == "final" { + break loop + } + case <-deadline: + require.Failf(t, "stress timed out", "delivered %d lines so far, want %d", + len(got), len(expected)) + } + } + + require.NoError(t, <-writeErr) + require.Equal(t, expected, got) +}