Settings watch: arm again when the kernel drops it (IN_IGNORED), said once; the test waits on the arm, not a sleep

go/buildstep/pull.go

@@ -102,6 +102,8 @@ type Puller struct {
 	// WatchSettings), and reads Capacity from it itself, keeping why it cannot be read (CapacityNote); a parked puller at
 	// capacity 0 therefore wakes when the node-engine refills the file, and a failed config follow is tried again then.
 	SettingsFile string
+	// watching, when set, is called each time the settings file's watch is (re)armed: a test waits on it, not on a sleep.
+	watching func()
 
 	mu      sync.Mutex
 	inHand  int

go/buildstep/pull_test.go

@@ -323,8 +323,16 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
 			}
 			return nil // followed; no value handed yet
 		}
+		armed := make(chan struct{}, 8)
+		var parked atomic.Int32
+		parks := func() int32 { return parked.Load() }
+		logf := func(format string, args ...any) {
+			if strings.Contains(format, "parks") {
+				parked.Add(1)
+			}
+		}
 		p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, PullWait: 10 * time.Millisecond,
-			SettingsFile: file, WatchConfig: watch,
+			SettingsFile: file, WatchConfig: watch, Logf: logf, watching: func() { armed <- struct{}{} },
 			Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -333,7 +341,7 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
 		if src.Waiting(delivery.AskDelivery) != 1 {
 			t.Fatal("a node whose capacity cannot be read took an ask")
 		}
-		time.Sleep(50 * time.Millisecond) // let the watch start
+		<-armed // the file watch is in place
 		tmp := file + ".new"
 		if err := os.WriteFile(tmp, []byte(`{"capacity":2}`), 0o600); err != nil {
 			t.Fatal(err)
@@ -342,6 +350,22 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
 			t.Fatal(err)
 		}
 		waitFor(t, "the ask taken once the file was refilled", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
+		// and the directory itself replaced (IN_IGNORED): the watch is armed again and a later refill still wakes the puller
+		dir := filepath.Dir(file)
+		before := parks()
+		if err := os.RemoveAll(dir); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Mkdir(dir, 0o700); err != nil {
+			t.Fatal(err)
+		}
+		<-armed // armed again
+		waitFor(t, "the puller parked with no settings file", func() bool { return parks() > before })
+		_ = src.PublishAsk(testAsk(delivery.AskDelivery))
+		if err := os.WriteFile(file, []byte(`{"capacity":2}`), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		waitFor(t, "the second ask taken after the directory was replaced", func() bool { a, _, _ := src.Fates(); return len(a) == 2 })
 		waitFor(t, "the failed follow tried again", func() bool { mu.Lock(); defer mu.Unlock(); return attempts >= 2 })
 	})
 	for _, c := range []struct {

go/buildstep/watch_settings_linux.go

@@ -10,9 +10,13 @@ import (
 	"syscall"
 )
 
+const settingsMask = syscall.IN_CLOSE_WRITE | syscall.IN_MOVED_TO | syscall.IN_CREATE
+
 // watchSettings wakes the puller each time its settings file is written or replaced (inotify on the file's directory, so
 // a rename over it counts): the event the node-engine's refill of the file is, which a parked puller at capacity 0
-// otherwise never sees. No timer: it blocks on the kernel's event. A watch that cannot start is said once.
+// otherwise never sees. No timer: it blocks on the kernel's event. When the kernel drops the watch (IN_IGNORED: the
+// directory was replaced or removed) it is added again, said once; while the directory is missing the nearest existing
+// ancestor is watched for the directory's return. A watch that cannot start is said once.
 func (p *Puller) watchSettings(ctx context.Context) {
 	if p.SettingsFile == "" {
 		return
@@ -25,10 +29,36 @@ func (p *Puller) watchSettings(ctx context.Context) {
 	f := os.NewFile(uintptr(fd), "inotify")
 	defer f.Close()
 	dir, base := filepath.Split(p.SettingsFile)
-	if _, err := syscall.InotifyAddWatch(fd, filepath.Clean(dir), syscall.IN_CLOSE_WRITE|syscall.IN_MOVED_TO|syscall.IN_CREATE); err != nil {
+	dir = filepath.Clean(dir)
+	// arm watches dir, or when it is missing its nearest existing ancestor; it returns the watch descriptor and whether it
+	// is the directory's own.
+	arm := func() (int, bool, error) {
+		wd, err := syscall.InotifyAddWatch(fd, dir, settingsMask)
+		if err == nil {
+			return wd, true, nil
+		}
+		for up := filepath.Dir(dir); ; up = filepath.Dir(up) {
+			wd, aerr := syscall.InotifyAddWatch(fd, up, settingsMask)
+			if aerr == nil {
+				// the directory may have come back between the two calls: look once more, now that the ancestor is watched
+				if own, oerr := syscall.InotifyAddWatch(fd, dir, settingsMask); oerr == nil {
+					return own, true, nil
+				}
+				return wd, false, nil
+			}
+			if up == filepath.Dir(up) {
+				return 0, false, err
+			}
+		}
+	}
+	wd, own, err := arm()
+	if err != nil {
 		p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
 		return
 	}
+	if p.watching != nil {
+		p.watching()
+	}
 	go func() { <-ctx.Done(); f.Close() }()
 	buf := make([]byte, 4096)
 	for {
@@ -36,8 +66,10 @@ func (p *Puller) watchSettings(ctx context.Context) {
 		if err != nil {
 			return
 		}
-		hit := false
+		hit, rearm := false, false
 		for off := 0; off+16 <= n; {
+			ewd := int(int32(binary.LittleEndian.Uint32(buf[off:])))
+			mask := binary.LittleEndian.Uint32(buf[off+4:])
 			l := int(binary.LittleEndian.Uint32(buf[off+12:]))
 			name := buf[off+16 : off+16+l]
 			for i, c := range name {
@@ -46,11 +78,33 @@ func (p *Puller) watchSettings(ctx context.Context) {
 					break
 				}
 			}
-			if string(name) == base {
+			switch {
+			case mask&syscall.IN_IGNORED != 0 && ewd == wd:
+				rearm = true
+			case ewd == wd && own && string(name) == base:
 				hit = true
+			case ewd == wd && !own:
+				rearm = true // something appeared where the directory was: look again
 			}
 			off += 16 + l
 		}
+		if rearm {
+			nwd, nown, err := arm()
+			if err != nil {
+				p.logf("the settings file %s is no longer watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
+				return
+			}
+			if own || nown { // said once per loss of the watch on the directory itself
+				if own {
+					p.logf("the watch of the settings file %s was dropped (its directory was replaced); watching again", p.SettingsFile)
+				}
+			}
+			wd, own = nwd, nown
+			if p.watching != nil {
+				p.watching()
+			}
+			hit = true // the file may have been refilled meanwhile
+		}
 		if hit {
 			p.Wake()
 		}