watch: fix stale watcher state across log rotation

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

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
+)

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=

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
 }
 

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() {

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
 			}
 		}
 	}()

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)
 	}
 

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)

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)
 }