From 254b9b0cd09434840510bdb4c147386e7a61cd27 Mon Sep 17 00:00:00 2001 From: Eugene Blikh Date: Thu, 6 Aug 2026 18:42:05 +0300 Subject: [PATCH] watch: fix stale watcher state across log rotation The recheck added in v1.4.14 arms the watch first and inspects the file afterwards, so a rotation can land on either side of the arming. Both sides left the tailer with a watcher it must not trust: * ChangeEvents seeded the watcher's size baseline with the caller's offset in the old file. When the name already pointed at the new, usually smaller file, the first write there looked like a truncation and the spurious reopen re-delivered every line already read from the new file. The baseline and the recorded inode now come from a stat of the file at arm time; changes that predate the watch are the recheck's job to find. * The pendingReopen and Truncated paths kept tail.changes across a reopen. When the rename fired right after the arming, the kept subscription was already dead and carried a latched Deleted notification, which triggered a bogus second reopen re-delivering everything read in between. Every reopen path now drops the subscription and re-arms it on the next EOF, and the recheck picks up whatever happened in the gap. Dropping is safe now that FileChanges.Stop shuts the producer goroutine down synchronously; an abandoned live producer would keep stealing events from the shared per-file channel. * A producer whose events channel was closed by an external RemoveWatch (e.g. Cleanup) exited silently and removed the watch a second time, driving the shared refcount negative: the tailer blocked forever, and every later watch of the same name skipped the fsnotify subscription and never got a single event. The producer now reports the file as deleted instead, and the tracker refuses to drive a refcount below zero. A stale pendingReopen is also cleared when a pending delete performs the reopen, so it cannot cause one more reopen cycle later. The new delayBeforeRecheck test knob mirrors delayBeforeWatch and widens the arm-to-recheck window for the regression tests. Part of TNTP-3131 --- CHANGES.md | 10 +++ go.mod | 8 +- go.sum | 10 +++ tail.go | 43 +++++++++- watch/filechanges.go | 30 ++++++- watch/inotify.go | 30 +++++-- watch/inotify_tracker.go | 16 ++-- watch/polling.go | 13 ++- watch/watch.go | 13 +-- watch_baseline_test.go | 100 ++++++++++++++++++++++ watch_lifecycle_test.go | 174 +++++++++++++++++++++++++++++++++++++++ 11 files changed, 422 insertions(+), 25 deletions(-) create mode 100644 watch_baseline_test.go create mode 100644 watch_lifecycle_test.go 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) +}