From 9e5666b4d6f581930fb0227ed65156dcca9aeb69 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:08:21 +0100 Subject: [PATCH] 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 | 2 + go/buildstep/pull_test.go | 28 ++++++++++++- go/buildstep/watch_settings_linux.go | 62 ++++++++++++++++++++++++++-- 3 files changed, 86 insertions(+), 6 deletions(-) diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 0474145..da1e540 100644 --- a/go/buildstep/pull.go +++ b/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 diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 30c6b66..24656f6 100644 --- a/go/buildstep/pull_test.go +++ b/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 { diff --git a/go/buildstep/watch_settings_linux.go b/go/buildstep/watch_settings_linux.go index f9d776d..5a01b96 100644 --- a/go/buildstep/watch_settings_linux.go +++ b/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() }