28f73e3 refactor(watch): drop the lost event flag and dead poll fallbacks · kvconf

M open_other.go +0 -3


@@ -5,8 +5,5 @@ package kvconf
5	5		import "os"
6	6
7	7		func openNonblocking(path string) (*os.File, error) {
8		-	if info, err := os.Stat(path); err == nil && !info.Mode().IsRegular() {
9		-	return nil, &os.PathError{Op: "open", Path: path, Err: ErrNotRegular}
10		-	}
11	8		return os.Open(path)
12	9		}


@@ -5,8 +5,5 @@ package kvconf
5		import "os"	5	import "os"
6			6
7		func openNonblocking(path string) (*os.File, error) {	7	func openNonblocking(path string) (*os.File, error) {
8	-	if info, err := os.Stat(path); err == nil && !info.Mode().IsRegular() {
9	-	return nil, &os.PathError{Op: "open", Path: path, Err: ErrNotRegular}
10	-	}
11		return os.Open(path)	8	return os.Open(path)
12		}	9	}

M watch.go +3 -11


@@ -68,15 +68,13 @@ func Watch(ctx context.Context, path string, opts WatchOptions, out chan<- struc
68	68		b = newPoller(opts.Poll)
69	69		} else if b, err = newNative(); err != nil {
70	70		return err
71		-	} else if b == nil {
72		-	b = newPoller(time.Second)
73	71		}
74	72		defer b.close() //nolint:errcheck // nothing to report once watching ends
75	73		stop := context.AfterFunc(ctx, b.wake)
76	74		defer stop()
77	75
78	76		w := &watcher{path: path, dir: opts.Dir, stamps: map[string]stamp{}}
79		-	missing, err := w.rewatch(b)
	77	+	missing, _, err := w.rewatchChanged(b)
80	78		if err != nil {
81	79		return err
82	80		}

@@ -189,10 +187,9 @@ func (t *target) has(name string) bool {
189	187
190	188		// event is a change in a watched directory. An empty name means "something
191	189		// changed in dir" (kqueue, polling, overflow); an empty dir means anything
192		-	// may have changed. lost means the watch itself vanished.
	190	+	// may have changed, or a watch was lost.
193	191		type event struct {
194	192		dir, name string
195		-	lost bool
196	193		// moved: the name was created, removed or renamed, so a link may now
197	194		// point elsewhere. A plain write leaves links alone.
198	195		moved bool

@@ -329,11 +326,6 @@ func (w watcher) plan() (targets []target, files []string) {
329	326		return targets, files
330	327		}
331	328
332		-	func (w *watcher) rewatch(b backend) (missing bool, err error) {
333		-	missing, _, err = w.rewatchChanged(b)
334		-	return missing, err
335		-	}
336		-
337	329		// rewatchChanged installs the watches again and reports whether the watched
338	330		// files changed meanwhile: a link was retargeted, a missing directory
339	331		// appeared, or a file changed while it was not watched.

@@ -367,7 +359,7 @@ func sameTargets(a, b []*target) bool {
367	359		func (w *watcher) classify(evs []event) (relevant, rewatch, recheck bool) {
368	360		relinked := false
369	361		for _, ev := range evs {
370		-	if ev.lost \|\| ev.dir == "" {
	362	+	if ev.dir == "" {
371	363		relevant, rewatch, recheck = true, true, true
372	364		continue
373	365		}

 		b = newPoller(opts.Poll)
 	} else if b, err = newNative(); err != nil {
 		return err
-	} else if b == nil {
-		b = newPoller(time.Second)
 	}
 	defer b.close() //nolint:errcheck // nothing to report once watching ends
 	stop := context.AfterFunc(ctx, b.wake)
 	defer stop()
 	w := &watcher{path: path, dir: opts.Dir, stamps: map[string]stamp{}}
-	missing, err := w.rewatch(b)
+	missing, _, err := w.rewatchChanged(b)
 	if err != nil {
 		return err
 	}
 // event is a change in a watched directory. An empty name means "something
 // changed in dir" (kqueue, polling, overflow); an empty dir means anything
-// may have changed. lost means the watch itself vanished.
+// may have changed, or a watch was lost.
 type event struct {
 	dir, name string
-	lost      bool
 	// moved: the name was created, removed or renamed, so a link may now
 	// point elsewhere. A plain write leaves links alone.
 	moved bool
 	return targets, files
 }
-func (w *watcher) rewatch(b backend) (missing bool, err error) {
-	missing, _, err = w.rewatchChanged(b)
-	return missing, err
-}
 // rewatchChanged installs the watches again and reports whether the watched
 // files changed meanwhile: a link was retargeted, a missing directory
 // appeared, or a file changed while it was not watched.
 func (w *watcher) classify(evs []event) (relevant, rewatch, recheck bool) {
 	relinked := false
 	for _, ev := range evs {
-		if ev.lost || ev.dir == "" {
+		if ev.dir == "" {
 			relevant, rewatch, recheck = true, true, true
 			continue
 		}

M watch_kqueue.go +6 -3

@@ -1,4 +1,4 @@
1		-	//go:build darwin \|\| ios \|\| freebsd \|\| netbsd \|\| openbsd \|\| dragonfly
	1	+	//go:build darwin \|\| freebsd \|\| netbsd \|\| openbsd \|\| dragonfly
2	2
3	3		package kvconf
4	4

@@ -124,8 +124,11 @@ func (k *kqueue) wait(timeout time.Duration, evs []event) ([]event, error) {
124	124		if !ok {
125	125		continue
126	126		}
127		-	lost := ev.Fflags&(syscall.NOTE_DELETE\|syscall.NOTE_RENAME\|syscall.NOTE_REVOKE) != 0
128		-	evs = append(evs, event{dir: dir, lost: lost})
	127	+	if ev.Fflags&(syscall.NOTE_DELETE\|syscall.NOTE_RENAME\|syscall.NOTE_REVOKE) != 0 {
	128	+	evs = append(evs, event{}) // a watch was lost
	129	+	continue
	130	+	}
	131	+	evs = append(evs, event{dir: dir})
129	132		}
130	133		return evs, nil
131	134		}

-//go:build darwin || ios || freebsd || netbsd || openbsd || dragonfly
+//go:build darwin || freebsd || netbsd || openbsd || dragonfly
 package kvconf
 		if !ok {
 			continue
 		}
-		lost := ev.Fflags&(syscall.NOTE_DELETE|syscall.NOTE_RENAME|syscall.NOTE_REVOKE) != 0
+		if ev.Fflags&(syscall.NOTE_DELETE|syscall.NOTE_RENAME|syscall.NOTE_REVOKE) != 0 {
-		evs = append(evs, event{dir: dir, lost: lost})
+			evs = append(evs, event{}) // a watch was lost
+			continue
+		}
+		evs = append(evs, event{dir: dir})
 	}
 	return evs, nil
 }

M watch_kqueue_darwin.go +1 -1

@@ -1,4 +1,4 @@
1		-	//go:build darwin \|\| ios
	1	+	//go:build darwin
2	2
3	3		package kvconf
4	4

@@ -1,4 +1,4 @@
1	-	//go:build darwin \|\| ios	1	+	//go:build darwin
2			2
3		package kvconf	3		package kvconf
4			4

M watch_linux.go +1 -1


@@ -123,7 +123,7 @@ func (in *inotify) wait(timeout time.Duration, evs []event) ([]event, error) {
123	123		if in.wds[dir] == ev.Wd {
124	124		delete(in.wds, dir)
125	125		}
126		-	evs = append(evs, event{dir: dir, lost: true})
	126	+	evs = append(evs, event{})
127	127		continue
128	128		}
129	129		for len(name) > 0 && name[len(name)-1] == 0 {


@@ -123,7 +123,7 @@ func (in *inotify) wait(timeout time.Duration, evs []event) ([]event, error) {
123		if in.wds[dir] == ev.Wd {	123		if in.wds[dir] == ev.Wd {
124		delete(in.wds, dir)	124		delete(in.wds, dir)
125		}	125		}
126	-	evs = append(evs, event{dir: dir, lost: true})	126	+	evs = append(evs, event{})
127		continue	127		continue
128		}	128		}
129		for len(name) > 0 && name[len(name)-1] == 0 {	129		for len(name) > 0 && name[len(name)-1] == 0 {

M watch_other.go +5 -3

@@ -1,6 +1,8 @@
1		-	//go:build !linux && !windows && !darwin && !ios && !freebsd && !netbsd && !openbsd && !dragonfly
	1	+	//go:build !linux && !windows && !darwin && !freebsd && !netbsd && !openbsd && !dragonfly
2	2
3	3		package kvconf
4	4
5		-	// newNative returns nil: this OS has no native backend, Watch polls.
6		-	func newNative() (backend, error) { return nil, nil }
	5	+	import "time"
	6	+
	7	+	// newNative polls every second: this OS has no native backend.
	8	+	func newNative() (backend, error) { return newPoller(time.Second), nil }

@@ -1,6 +1,8 @@
1	-	//go:build !linux && !windows && !darwin && !ios && !freebsd && !netbsd && !openbsd && !dragonfly	1	+	//go:build !linux && !windows && !darwin && !freebsd && !netbsd && !openbsd && !dragonfly
2			2
3		package kvconf	3		package kvconf
4			4
5	-	// newNative returns nil: this OS has no native backend, Watch polls.	5	+	import "time"
6	-	func newNative() (backend, error) { return nil, nil }	6	+
			7	+	// newNative polls every second: this OS has no native backend.
			8	+	func newNative() (backend, error) { return newPoller(time.Second), nil }

M watch_test.go +47 -29


@@ -44,12 +44,23 @@ func (e *watchEnv) symlink(target, rel string) {
44	44		}
45	45		}
46	46
	47	+	// fastWait is how long a change that needs no retry may take. It stays below
	48	+	// retryMissing, so a broken fast path is not hidden by the periodic look.
	49	+	const fastWait = retryMissing - 500*time.Millisecond
	50	+
47	51		func (e *watchEnv) expectChange(why string) {
	52	+	e.t.Helper()
	53	+	e.expectChangeWithin(why, fastWait)
	54	+	}
	55	+
	56	+	// expectChangeWithin is expectChange for a step that legitimately relies on
	57	+	// the retry or recheck, which fires after retryMissing.
	58	+	func (e *watchEnv) expectChangeWithin(why string, d time.Duration) {
48	59		e.t.Helper()
49	60		select {
50	61		case <-e.out:
51		-	case <-time.After(5 * time.Second):
52		-	e.t.Fatalf("no change reported after %s", why)
	62	+	case <-time.After(d):
	63	+	e.t.Fatalf("no change reported within %v after %s", d, why)
53	64		}
54	65		}
55	66

@@ -76,6 +87,14 @@ type backendCase struct {
76	87		poll time.Duration
77	88		}
78	89
	90	+	// slowWait is the deadline for a step that relies on the retry on this OS.
	91	+	func slowWait() time.Duration {
	92	+	if runtime.GOOS == "windows" {
	93	+	return retryMissing + 3*time.Second
	94	+	}
	95	+	return fastWait
	96	+	}
	97	+
79	98		func backends() []backendCase {
80	99		return []backendCase{{name: "native"}, {name: "poll", poll: 20 * time.Millisecond}}
81	100		}

@@ -135,7 +154,8 @@ func TestWatch(t *testing.T) {
135	154		rel: "a/b/config",
136	155		steps: func(e *watchEnv) {
137	156		e.write("a/b/config", "a = 1\n")
138		-	e.expectChange("the directories and file appeared")
	157	+	// Windows may only find the new directories on the retry.
	158	+	e.expectChangeWithin("the directories and file appeared", slowWait())
139	159		e.settle()
140	160		e.write("a/b/config", "a = 2\n")
141	161		e.expectChange("a write once the directory exists")

@@ -147,10 +167,12 @@ func TestWatch(t *testing.T) {
147	167		setup: func(e *watchEnv) { e.write("d/config", "a = 1\n") },
148	168		steps: func(e *watchEnv) {
149	169		must(e.t, os.RemoveAll(filepath.Join(e.dir, "d")))
150		-	e.expectChange("the directory removal")
	170	+	// Windows reports a removed directory only after the watch on
	171	+	// it is closed: the recheck finds it.
	172	+	e.expectChangeWithin("the directory removal", slowWait())
151	173		e.settle()
152	174		e.write("d/config", "a = 2\n")
153		-	e.expectChange("the recreated file")
	175	+	e.expectChangeWithin("the recreated file", slowWait())
154	176		},
155	177		},
156	178		{

@@ -286,9 +308,17 @@ func TestWatch(t *testing.T) {
286	308		}
287	309		}
288	310
	311	+	// Every write restarts the debounce: a burst longer than the debounce is
	312	+	// reported once, after its last write.
289	313		func TestWatchDebounce(t *testing.T) {
	314	+	const (
	315	+	debounce = 200 * time.Millisecond
	316	+	writes = 10
	317	+	gap = 50 * time.Millisecond
	318	+	)
290	319		for _, bc := range backends() {
291	320		t.Run(bc.name, func(t *testing.T) {
	321	+	t.Parallel()
292	322		dir := t.TempDir()
293	323		path := filepath.Join(dir, "config")
294	324		must(t, os.WriteFile(path, nil, 0o600))

@@ -297,15 +327,20 @@ func TestWatchDebounce(t *testing.T) {
297	327		defer cancel()
298	328		ready := make(chan struct{})
299	329		go func() {
300		-	_ = Watch(ctx, path, WatchOptions{Poll: bc.poll, Debounce: 300 * time.Millisecond, Ready: ready}, out)
	330	+	_ = Watch(ctx, path, WatchOptions{Poll: bc.poll, Debounce: debounce, Ready: ready}, out)
301	331		}()
302	332		<-ready
303		-	for i := range 5 {
	333	+	var lastWrite time.Time
	334	+	for i := range writes {
	335	+	lastWrite = time.Now()
304	336		must(t, os.WriteFile(path, []byte{byte('a' + i)}, 0o600))
305		-	time.Sleep(50 * time.Millisecond)
	337	+	time.Sleep(gap)
306	338		}
307	339		select {
308	340		case <-out:
	341	+	if d := time.Since(lastWrite); d < debounce {
	342	+	t.Fatalf("reported %v after the last write, before the %v debounce", d, debounce)
	343	+	}
309	344		case <-time.After(5 * time.Second):
310	345		t.Fatal("no change reported")
311	346		}

@@ -319,10 +354,12 @@ func TestWatchDebounce(t *testing.T) {
319	354		}
320	355
321	356		// An atomic save is one change: nothing is reported again when the watcher
322		-	// looks at the files later.
	357	+	// looks at the files later. This guards Windows, where the recheck after a
	358	+	// rename would otherwise report the same save a second time.
323	359		func TestWatchReportsASaveOnce(t *testing.T) {
324	360		for _, bc := range backends() {
325	361		t.Run(bc.name, func(t *testing.T) {
	362	+	t.Parallel()
326	363		dir := t.TempDir()
327	364		path := filepath.Join(dir, "config")
328	365		must(t, os.WriteFile(path, []byte("a = 1\n"), 0o600))

@@ -337,7 +374,7 @@ func TestWatchReportsASaveOnce(t *testing.T) {
337	374		must(t, WriteFile(path, []byte("a = 2\n"), 0o600))
338	375		select {
339	376		case <-out:
340		-	case <-time.After(5 * time.Second):
	377	+	case <-time.After(slowWait()):
341	378		t.Fatal("no change reported")
342	379		}
343	380		select {

@@ -348,22 +385,3 @@ func TestWatchReportsASaveOnce(t *testing.T) {
348	385		})
349	386		}
350	387		}
351		-
352		-	func TestWatchCancelBeforeEvents(t *testing.T) {
353		-	for _, bc := range backends() {
354		-	t.Run(bc.name, func(t *testing.T) {
355		-	ctx, cancel := context.WithCancel(context.Background())
356		-	cancel()
357		-	done := make(chan error, 1)
358		-	go func() {
359		-	done <- Watch(ctx, filepath.Join(t.TempDir(), "config"), WatchOptions{Poll: bc.poll}, make(chan struct{}, 1))
360		-	}()
361		-	select {
362		-	case err := <-done:
363		-	must(t, err)
364		-	case <-time.After(5 * time.Second):
365		-	t.Fatal("Watch ignored a cancelled context")
366		-	}
367		-	})
368		-	}
369		-	}

 	}
 }
+// fastWait is how long a change that needs no retry may take. It stays below
+// retryMissing, so a broken fast path is not hidden by the periodic look.
+const fastWait = retryMissing - 500*time.Millisecond
 func (e *watchEnv) expectChange(why string) {
+	e.t.Helper()
+	e.expectChangeWithin(why, fastWait)
+}
+// expectChangeWithin is expectChange for a step that legitimately relies on
+// the retry or recheck, which fires after retryMissing.
+func (e *watchEnv) expectChangeWithin(why string, d time.Duration) {
 	e.t.Helper()
 	select {
 	case <-e.out:
-	case <-time.After(5 * time.Second):
+	case <-time.After(d):
-		e.t.Fatalf("no change reported after %s", why)
+		e.t.Fatalf("no change reported within %v after %s", d, why)
 	}
 }
 	poll time.Duration
 }
+// slowWait is the deadline for a step that relies on the retry on this OS.
+func slowWait() time.Duration {
+	if runtime.GOOS == "windows" {
+		return retryMissing + 3*time.Second
+	}
+	return fastWait
+}
 func backends() []backendCase {
 	return []backendCase{{name: "native"}, {name: "poll", poll: 20 * time.Millisecond}}
 }
 			rel:  "a/b/config",
 			steps: func(e *watchEnv) {
 				e.write("a/b/config", "a = 1\n")
-				e.expectChange("the directories and file appeared")
+				// Windows may only find the new directories on the retry.
+				e.expectChangeWithin("the directories and file appeared", slowWait())
 				e.settle()
 				e.write("a/b/config", "a = 2\n")
 				e.expectChange("a write once the directory exists")
 			setup: func(e *watchEnv) { e.write("d/config", "a = 1\n") },
 			steps: func(e *watchEnv) {
 				must(e.t, os.RemoveAll(filepath.Join(e.dir, "d")))
-				e.expectChange("the directory removal")
+				// Windows reports a removed directory only after the watch on
+				// it is closed: the recheck finds it.
+				e.expectChangeWithin("the directory removal", slowWait())
 				e.settle()
 				e.write("d/config", "a = 2\n")
-				e.expectChange("the recreated file")
+				e.expectChangeWithin("the recreated file", slowWait())
 			},
 		},
 		{
 	}
 }
+// Every write restarts the debounce: a burst longer than the debounce is
+// reported once, after its last write.
 func TestWatchDebounce(t *testing.T) {
+	const (
+		debounce = 200 * time.Millisecond
+		writes   = 10
+		gap      = 50 * time.Millisecond
+	)
 	for _, bc := range backends() {
 		t.Run(bc.name, func(t *testing.T) {
+			t.Parallel()
 			dir := t.TempDir()
 			path := filepath.Join(dir, "config")
 			must(t, os.WriteFile(path, nil, 0o600))
 			defer cancel()
 			ready := make(chan struct{})
 			go func() {
-				_ = Watch(ctx, path, WatchOptions{Poll: bc.poll, Debounce: 300 * time.Millisecond, Ready: ready}, out)
+				_ = Watch(ctx, path, WatchOptions{Poll: bc.poll, Debounce: debounce, Ready: ready}, out)
 			}()
 			<-ready
-			for i := range 5 {
+			var lastWrite time.Time
+			for i := range writes {
+				lastWrite = time.Now()
 				must(t, os.WriteFile(path, []byte{byte('a' + i)}, 0o600))
-				time.Sleep(50 * time.Millisecond)
+				time.Sleep(gap)
 			}
 			select {
 			case <-out:
+				if d := time.Since(lastWrite); d < debounce {
+					t.Fatalf("reported %v after the last write, before the %v debounce", d, debounce)
+				}
 			case <-time.After(5 * time.Second):
 				t.Fatal("no change reported")
 			}
 }
 // An atomic save is one change: nothing is reported again when the watcher
-// looks at the files later.
+// looks at the files later. This guards Windows, where the recheck after a
+// rename would otherwise report the same save a second time.
 func TestWatchReportsASaveOnce(t *testing.T) {
 	for _, bc := range backends() {
 		t.Run(bc.name, func(t *testing.T) {
+			t.Parallel()
 			dir := t.TempDir()
 			path := filepath.Join(dir, "config")
 			must(t, os.WriteFile(path, []byte("a = 1\n"), 0o600))
 			must(t, WriteFile(path, []byte("a = 2\n"), 0o600))
 			select {
 			case <-out:
-			case <-time.After(5 * time.Second):
+			case <-time.After(slowWait()):
 				t.Fatal("no change reported")
 			}
 			select {
 		})
 	}
 }
-func TestWatchCancelBeforeEvents(t *testing.T) {
-	for _, bc := range backends() {
-		t.Run(bc.name, func(t *testing.T) {
-			ctx, cancel := context.WithCancel(context.Background())
-			cancel()
-			done := make(chan error, 1)
-			go func() {
-				done <- Watch(ctx, filepath.Join(t.TempDir(), "config"), WatchOptions{Poll: bc.poll}, make(chan struct{}, 1))
-			}()
-			select {
-			case err := <-done:
-				must(t, err)
-			case <-time.After(5 * time.Second):
-				t.Fatal("Watch ignored a cancelled context")
-			}
-		})
-	}
-}

M watch_windows.go +4 -7


@@ -8,8 +8,8 @@ import (
8	8		"unsafe"
9	9		)
10	10
11		-	// The standard syscall package lacks these, or declares them with a wrong
12		-	// signature on 64-bit Windows; they are loaded from kernel32 directly.
	11	+	// These functions are missing from package syscall: they are loaded from
	12	+	// kernel32 directly.
13	13		var (
14	14		kernel32 = syscall.NewLazyDLL("kernel32.dll")
15	15		procCreateEventW = kernel32.NewProc("CreateEventW")

@@ -58,9 +58,6 @@ func createEvent() (syscall.Handle, error) {
58	58		}
59	59
60	60		func newNative() (backend, error) {
61		-	if err := procWaitForMultipleObjects.Find(); err != nil {
62		-	return nil, nil // poll instead
63		-	}
64	61		ev, err := createEvent()
65	62		if err != nil {
66	63		return nil, err

@@ -186,7 +183,7 @@ func (w *winWatch) wait(timeout time.Duration, evs []event) ([]event, error) {
186	183		return evs, nil
187	184		}
188	185		// The directory was removed or became unreadable.
189		-	return append(evs, event{dir: d.dir, lost: true}), nil
	186	+	return append(evs, event{}), nil
190	187		}
191	188		if n == 0 {
192	189		evs = append(evs, event{dir: d.dir}) // the buffer overflowed

@@ -203,7 +200,7 @@ func (w *winWatch) wait(timeout time.Duration, evs []event) ([]event, error) {
203	200		off += info.NextEntryOffset
204	201		}
205	202		if err := d.read(); err != nil {
206		-	evs = append(evs, event{dir: d.dir, lost: true})
	203	+	evs = append(evs, event{})
207	204		}
208	205		return evs, nil
209	206		}

 	"unsafe"
 )
-// The standard syscall package lacks these, or declares them with a wrong
+// These functions are missing from package syscall: they are loaded from
-// signature on 64-bit Windows; they are loaded from kernel32 directly.
+// kernel32 directly.
 var (
 	kernel32                   = syscall.NewLazyDLL("kernel32.dll")
 	procCreateEventW           = kernel32.NewProc("CreateEventW")
 }
 func newNative() (backend, error) {
-	if err := procWaitForMultipleObjects.Find(); err != nil {
-		return nil, nil // poll instead
-	}
 	ev, err := createEvent()
 	if err != nil {
 		return nil, err
 			return evs, nil
 		}
 		// The directory was removed or became unreadable.
-		return append(evs, event{dir: d.dir, lost: true}), nil
+		return append(evs, event{}), nil
 	}
 	if n == 0 {
 		evs = append(evs, event{dir: d.dir}) // the buffer overflowed
 		off += info.NextEntryOffset
 	}
 	if err := d.read(); err != nil {
-		evs = append(evs, event{dir: d.dir, lost: true})
+		evs = append(evs, event{})
 	}
 	return evs, nil
 }