From 769a8d6f6a3c88815452acd66c711c2236375bff Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 21:45:13 +0200 Subject: [PATCH 1/2] Fill again what failed for want of the node-engine's write socket, when inotify says the socket is back (issue 1165) --- .../cmd/mesh-provisioner-client/client.go | 38 +++++- .../mesh-provisioner-client/client_test.go | 117 +++++++++++++++++- .../cmd/mesh-provisioner-client/main.go | 6 +- .../mesh-provisioner-client/socketwatch.go | 103 +++++++++++++++ modules/mesh-provisioner-client/go.mod | 2 +- 5 files changed, 260 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go diff --git a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go index 52ea5bd4..7bd28b50 100644 --- a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go +++ b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go @@ -119,6 +119,9 @@ type Client struct { Templates *templateStore // Now dates the reports; nil is time.Now. Now func() time.Time + // SocketUp wakes the client when the node-engine's write socket is made again (watchSocket); nil never does. + // A module whose write failed for want of the socket is filled again then (issue 1165), and nothing else is. + SocketUp <-chan struct{} // watchBound watches a provider's bucket of entries for this machine's keys, once per run (consumers.go); set by // Run. @@ -147,6 +150,8 @@ type state struct { // of the consumers left out of one of its files. providing map[string]bool leftOut map[string]string + // unplaced is a module whose last fill failed for want of the node-engine's write socket (issue 1165). + unplaced map[string]bool // bound is each provider's bucket of entries this client watches, by key: this machine's consumers' entries alone; // boundReady is a bucket whose current entries are all read, and boundUsers the modules filled from each. bound map[string]map[string][]byte @@ -161,6 +166,8 @@ type update struct { changed []byte // asked is the tool runner asking for a module's value (asks.go). Set, entry is nil. asked *nats.Msg + // socketUp is the node-engine's write socket made again (issue 1165). Set, entry is nil. + socketUp bool } // Run serves until ctx ends or a watch stops, which it returns: run again, it starts from the current values. @@ -248,6 +255,23 @@ func (c *Client) Run(ctx context.Context) error { return fmt.Errorf("listening for %s: %w", SecretChangedSubject, err) } defer func() { _ = said.Unsubscribe() }() + // The node-engine's write socket made again: what failed for want of it is filled again, in the loop below. + if c.SocketUp != nil { + go func() { + for { + select { + case <-ctx.Done(): + return + case <-c.SocketUp: + select { + case updates <- update{socketUp: true}: + case <-ctx.Done(): + return + } + } + } + }() + } // This machine's tool runner asking for a module's config value (asks.go): answered from this run's state, in // its loop, so an answer is never read from half an update. asks, err := c.Conn.Subscribe(AskSubject(c.Node), func(m *nats.Msg) { @@ -266,7 +290,7 @@ func (c *Client) Run(ctx context.Context) error { reported: map[string][]byte{}, secrets: map[string]heldSecret{}, items: map[string]map[string]itemValue{}, recorded: map[string][]byte{}, assigned: map[string]bool{}, binds: map[string]binding{}, providing: map[string]bool{}, leftOut: map[string]string{}, bound: map[string]map[string][]byte{}, - boundReady: map[string]bool{}, boundUsers: map[string]map[string]bool{}} + boundReady: map[string]bool{}, unplaced: map[string]bool{}, boundUsers: map[string]map[string]bool{}} // Every current value first: nothing is filled from half of them. current := map[string]bool{} var early []*nats.Msg @@ -282,6 +306,9 @@ func (c *Client) Run(ctx context.Context) error { early = append(early, u.asked) continue } + if u.socketUp { + continue // the first fill below tries every file + } if u.changed != nil { // Nothing is held yet: every secret is asked for when it is first filled. continue @@ -326,6 +353,10 @@ func (c *Client) Run(ctx context.Context) error { switch { case u.asked != nil: c.respond(s, u.asked) + case u.socketUp: + for m := range s.unplaced { + dirty[m] = true + } case u.changed != nil: for _, m := range c.secretChanged(s, u.changed) { dirty[m] = true @@ -550,6 +581,7 @@ func (c *Client) leftModules(s *state) []string { // leave removes a module's needs and its report: it is no longer the provisioner's on this node. func (c *Client) leave(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string) { prefix := c.Node + "." + module + "." + delete(s.unplaced, module) removed := 0 for _, key := range sortedKeys(s.needs) { if strings.HasPrefix(key, prefix) { @@ -660,6 +692,7 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu c.forgetItems(ctx, kvs, s, module, keep) } moved := false + delete(s.unplaced, module) for _, f := range targets { // File by file (issue 1553, ADR 0349 clause 4): a target with a value that waits keeps the file there is. if !allKnown(f.text, values) { @@ -672,6 +705,9 @@ func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValu err = errFill(a) } if err != nil { + if socketAbsent(err) { + s.unplaced[module] = true + } failed = append(failed, fmt.Sprintf("%s: %v", f.Path, err)) continue } diff --git a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go index 342ecb2d..fc71fca8 100644 --- a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go +++ b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go @@ -200,6 +200,14 @@ type engine struct { } func startEngine(t *testing.T, owners map[string]string) *engine { + t.Helper() + e := newEngine(t, owners) + e.listen(t) + return e +} + +// newEngine is an engine whose socket is not made yet: a node-engine not started, or replacing itself. +func newEngine(t *testing.T, owners map[string]string) *engine { t.Helper() // A unix socket's path is short: under /tmp, not the test's own directory. dir, err := os.MkdirTemp("/tmp", "p5-") @@ -209,6 +217,12 @@ func startEngine(t *testing.T, owners map[string]string) *engine { t.Cleanup(func() { os.RemoveAll(dir) }) e := &engine{root: t.TempDir(), socket: filepath.Join(dir, "fill.sock"), owner: owners, moved: map[string]int{}, asked: map[string]int{}, wrote: map[string]string{}} + return e +} + +// listen makes the socket and answers on it until stop, which removes it as a node-engine stopping does. +func (e *engine) listen(t *testing.T) (stop func()) { + t.Helper() l, err := net.Listen("unix", e.socket) if err != nil { t.Fatal(err) @@ -233,7 +247,7 @@ func startEngine(t *testing.T, owners map[string]string) *engine { }() } }() - return e + return func() { l.Close(); os.Remove(e.socket) } } func (e *engine) at(path string) string { return filepath.Join(e.root, path) } @@ -349,10 +363,14 @@ func aClient(t *testing.T, b *aBus, e *engine, state string) (*Client, func()) { t.Fatal(err) } secrets.Within = 2 * time.Second - c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets, + ctx, cancel := context.WithCancel(context.Background()) + up, err := watchSocket(ctx, e.socket) + if err != nil { + t.Fatal(err) + } + c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets, SocketUp: up, Fill: SocketFiller{Socket: e.socket, Busy: 10 * time.Millisecond, Tries: 3}, Templates: &templateStore{dir: state, fetch: HTTPFetch}} - ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- c.Run(ctx) }() stopped := false @@ -703,3 +721,96 @@ func TestNatsResticsFilesFillFromTheBackupPlacement(t *testing.T) { b.assign(t, "restic", plan("# a\n/srv/a\n", "# a\nstate\n")) eventually(t, "a module taken away refills backups.conf", func() bool { return e.read(t, backups) == head+"# a\n/srv/a\n" }) } + +// Issue 1165: a file whose write failed because the node-engine's socket was gone is written when the socket is made +// again, with no timer; a file that was placed is not asked again by it. +func TestNatsAFillThatFailedForWantOfTheSocketIsTriedWhenTheSocketReturns(t *testing.T) { + b := startBus(t) + heard := b.listen(t) + e := newEngine(t, map[string]string{dunstrc: "owned"}) + b.config(t, "dunst", "font-size", "13") + b.assign(t, "dunst", dunst(t)) + aClient(t, b, e, t.TempDir()) + + eventually(t, "the failure is said", func() bool { + r, _ := heard.last("dunst") + return r.Status == StatusFailed && strings.Contains(r.Reason, "fill.sock") + }) + if got := e.read(t, dunstrc); got != "" { + t.Fatalf("written without a socket: %q", got) + } + e.listen(t) + eventually(t, "dunstrc is written once the socket is back", func() bool { + return strings.Contains(e.read(t, dunstrc), "Monospace 13") + }) + eventually(t, "and the report says applied", func() bool { + r, _ := heard.last("dunst") + return r.Status == StatusApplied + }) +} + +func TestNatsAReturningSocketAsksNothingOfAModuleThatWasPlaced(t *testing.T) { + b := startBus(t) + e := startEngine(t, map[string]string{dunstrc: "owned"}) + b.config(t, "dunst", "font-size", "13") + b.assign(t, "dunst", dunst(t)) + c, _ := aClient(t, b, e, t.TempDir()) + eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") }) + _, before := e.counts(dunstrc) + + // A socket-up wake with nothing unplaced, then a later change as the proof the loop has run past it. + up := make(chan struct{}, 1) + c.SocketUp = up + _ = up + b.config(t, "dunst", "font-size", "15") + eventually(t, "the change is written", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 15") }) + _, after := e.counts(dunstrc) + if after-before != 1 { + t.Errorf("dunstrc was asked %d times for one change", after-before) + } +} + +func TestWatchSocketWakesWhenTheSocketIsMadeAndWhenItsDirectoryReturns(t *testing.T) { + base, err := os.MkdirTemp("/tmp", "ws-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(base) }) + dir := filepath.Join(base, "engine") + if err := os.Mkdir(dir, 0o755); err != nil { + t.Fatal(err) + } + socket := filepath.Join(dir, "fill.sock") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + up, err := watchSocket(ctx, socket) + if err != nil { + t.Fatal(err) + } + heard := func(what string) { + t.Helper() + select { + case <-up: + case <-time.After(5 * time.Second): + t.Fatalf("no wake for %s", what) + } + } + l, err := net.Listen("unix", socket) + if err != nil { + t.Fatal(err) + } + heard("the socket made") + l.Close() + // The whole directory goes, as with a runtime directory removed on stop, and returns with the socket. + if err := os.RemoveAll(dir); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(dir, 0o755); err != nil { + t.Fatal(err) + } + if l, err = net.Listen("unix", socket); err != nil { + t.Fatal(err) + } + defer l.Close() + heard("the directory and socket back") +} diff --git a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go index 72919bc8..3d9dc49c 100644 --- a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go +++ b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go @@ -133,7 +133,11 @@ func serve(ctx context.Context, credential, socket, stateDir string) error { if err != nil { return err } - client := &Client{Node: c.Node, JS: js, Conn: nc, Secrets: secrets, + up, err := watchSocket(ctx, socket) + if err != nil { + return err + } + client := &Client{Node: c.Node, SocketUp: up, JS: js, Conn: nc, Secrets: secrets, Fill: SocketFiller{Socket: socket, Busy: time.Second, Tries: 10}, Templates: &templateStore{dir: filepath.Join(stateDir, "templates"), fetch: HTTPFetch}} wait := time.Second diff --git a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go new file mode 100644 index 00000000..78df0b6f --- /dev/null +++ b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go @@ -0,0 +1,103 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "syscall" + "unsafe" + + "golang.org/x/sys/unix" +) + +// socketAbsent says whether a write failed because nothing answers at the node-engine's write socket: its file is +// gone, or it is there with no one listening. That is the node-engine replacing itself (issue 1165); the write is +// tried again when the socket is back (watchSocket). +func socketAbsent(err error) bool { + return errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ECONNREFUSED) +} + +// watchSocket tells, on the channel it returns, each time the file at socket is made, moved in or changed in its +// attributes — a node-engine that starts makes its write socket and sets its mode — until ctx ends. It is the +// operating system's own event (inotify), not a timer: the client is on the node-engine's machine. Events that +// come while one is unread are one. When the socket's directory is gone it watches the directory above for its +// return. The channel only wakes: what is owed a write is the client's to read. +func watchSocket(ctx context.Context, socket string) (<-chan struct{}, error) { + fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK) + if err != nil { + return nil, fmt.Errorf("watching %s: %w", socket, err) + } + f := os.NewFile(uintptr(fd), "inotify") + out := make(chan struct{}, 1) + dir := filepath.Dir(socket) + // The directory's watch and, while the directory is absent, its parent's. + if err := addWatches(fd, dir); err != nil { + _ = f.Close() + return nil, fmt.Errorf("watching %s: %w", socket, err) + } + go func() { <-ctx.Done(); _ = f.Close() }() + go func() { + buf := make([]byte, 16*(unix.SizeofInotifyEvent+unix.NAME_MAX+1)) + for { + n, err := f.Read(buf) + if err != nil { + return + } + wake, rewatch := false, false + for off := 0; off+unix.SizeofInotifyEvent <= n; { + ev := (*unix.InotifyEvent)(unsafe.Pointer(&buf[off])) + name := "" + if ev.Len > 0 { + raw := buf[off+unix.SizeofInotifyEvent : off+unix.SizeofInotifyEvent+int(ev.Len)] + for i, b := range raw { + if b == 0 { + raw = raw[:i] + break + } + } + name = string(raw) + } + off += unix.SizeofInotifyEvent + int(ev.Len) + switch { + case ev.Mask&(unix.IN_DELETE_SELF|unix.IN_MOVE_SELF|unix.IN_IGNORED) != 0: + rewatch = true + case name == filepath.Base(socket): + wake = true + case name == filepath.Base(dir): + // The directory came back: watched again, and the socket may already be in it. + rewatch, wake = true, true + } + } + if rewatch { + if err := addWatches(fd, dir); err != nil { + logf("[provisioner-client] watching %s again: %v", socket, err) + } + if _, err := os.Lstat(socket); err == nil { + wake = true + } + } + if wake { + select { + case out <- struct{}{}: + default: + } + } + } + }() + return out, nil +} + +// addWatches watches dir for what is made in it, and its parent for dir's own return; a watch already there is kept. +func addWatches(fd int, dir string) error { + if _, err := unix.InotifyAddWatch(fd, filepath.Dir(dir), unix.IN_CREATE|unix.IN_MOVED_TO); err != nil { + return err + } + _, err := unix.InotifyAddWatch(fd, dir, unix.IN_CREATE|unix.IN_MOVED_TO|unix.IN_ATTRIB|unix.IN_DELETE_SELF| + unix.IN_MOVE_SELF) + if errors.Is(err, syscall.ENOENT) { + return nil // absent: its parent's watch tells when it is made + } + return err +} diff --git a/modules/mesh-provisioner-client/go.mod b/modules/mesh-provisioner-client/go.mod index 1d3e1806..c444be2e 100644 --- a/modules/mesh-provisioner-client/go.mod +++ b/modules/mesh-provisioner-client/go.mod @@ -6,6 +6,7 @@ require ( git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021114014-66e8fdaa62e2 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 + golang.org/x/sys v0.48.0 ) require ( @@ -17,6 +18,5 @@ require ( github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect - golang.org/x/sys v0.48.0 // indirect golang.org/x/time v0.15.0 // indirect ) From aa5d263c085f296419934b8b3f1aa956852e76ce Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 21:45:34 +0200 Subject: [PATCH 2/2] Test the returning socket by a real restart of the stand-in node-engine --- .../cmd/mesh-provisioner-client/client_test.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go index fc71fca8..32067281 100644 --- a/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go +++ b/modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go @@ -751,17 +751,17 @@ func TestNatsAFillThatFailedForWantOfTheSocketIsTriedWhenTheSocketReturns(t *tes func TestNatsAReturningSocketAsksNothingOfAModuleThatWasPlaced(t *testing.T) { b := startBus(t) - e := startEngine(t, map[string]string{dunstrc: "owned"}) + e := newEngine(t, map[string]string{dunstrc: "owned"}) + stop := e.listen(t) b.config(t, "dunst", "font-size", "13") b.assign(t, "dunst", dunst(t)) - c, _ := aClient(t, b, e, t.TempDir()) + aClient(t, b, e, t.TempDir()) eventually(t, "dunstrc is filled", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 13") }) _, before := e.counts(dunstrc) - // A socket-up wake with nothing unplaced, then a later change as the proof the loop has run past it. - up := make(chan struct{}, 1) - c.SocketUp = up - _ = up + // The node-engine replaces itself: the socket goes and returns, and a later change proves the loop ran past it. + stop() + e.listen(t) b.config(t, "dunst", "font-size", "15") eventually(t, "the change is written", func() bool { return strings.Contains(e.read(t, dunstrc), "Monospace 15") }) _, after := e.counts(dunstrc)