Ask no check of a group member that waits on its composed check, and stop asking a merge refused the same way with no trunk move (issue 1106)

modules/gitea/merge.ts

@@ -12,7 +12,7 @@ import { mergeableUnder, newestByContext } from "./checks.js";
 import { MERGE_METHODS, Refused, checkName } from "./guard.js";
 import { CHECK_CONTEXT, REPO_CHECK_CONTEXT } from "./pulls.js";
 import { approvedHeadsOf, groupRefusal, memberKey, namesSiblings, notApproved, readSiblings, siblingOutOfDate, siblingsFile } from "./siblings.js";
-import { announcementOf, askedWords, outOfDateWords, sameCommit, tookFile, tookWords, twelve, verdictAtTrunk } from "./trunk.js";
+import { announcementOf, askedWords, outOfDateWords, sameCommit, tookFile, tookWords, twelve, verdictAtTrunk, waitsForComposed } from "./trunk.js";
 
 /** A head named for a merge: at least 12 hex characters of its sha (novox/hq issue 381), up to a sha-256 commit's 64. */
 const REVIEWED_HEAD = /^[0-9a-f]{12,64}$/;
@@ -117,6 +117,14 @@ export function mergeTool(gitea: GiteaClient, given: MergeDeps): ToolDefinition
           : at.judged ? `was judged against trunk ${twelve(at.judged)}` : "names no trunk commit it was judged against";
         const took = tookWords(tookFile(deps.stateDir), `${owner}/${repo}`);
         const head = twelve(pull.head_sha), now = twelve(trunkHead);
+        // **A member that waits for its group's composed check is not announced again** (novox/hq issue 1106): its own
+        // check can only say "composed" once more, so an ask here brought a verdict naming no trunk and the next merge was
+        // refused the same way for ever. The composed check is mesh-delivery's to ask; this says whose verdict is awaited.
+        if (waitsForComposed(checks[CHECK_CONTEXT])) {
+          throw new Refused(`${owner}/${repo}#${number} is not merged: its merge gate ${checks[CHECK_CONTEXT]?.description}; the ` +
+            `verdict that names its trunk commit is its group's composed check's, and mesh-delivery asks it, this tool asks no check ` +
+            `(main is now ${now}). Ask again once the group's composed check has passed. Nothing was merged`);
+        }
         if (at.onItsWay) {
           throw new Refused(`${owner}/${repo}#${number} is not merged: its verdict ${judged}, and ${pull.base} is now ` +
             `${now}. Its check is already asked and not yet answered (${checks[CHECK_CONTEXT]?.description}), so it is not ` +

modules/gitea/pulls.ts

@@ -226,6 +226,10 @@ function repoWrong(repo: Layer | undefined, facts: RepoCheckFacts): boolean {
   return !!repo && repoStatus(repo, facts).state !== "success";
 }
 
+/** What a group member's gate says while the group's composed check is the one that judges it: the member's own check can say
+ *  nothing more of it, so asking that check again brings no verdict that names a trunk (novox/hq issue 1106). */
+export const WAITS_FOR_COMPOSED = "waits for its group's composed check";
+
 /** The forge's status for the gate. */
 export function statusFor(c: Checked): CommitStatus {
   const gate = c.gate ?? { verdict: c.verdict, summary: c.summary };
@@ -260,7 +264,7 @@ export function statusesFor(c: Checked, facts: RepoCheckFacts = {}, gateNow?: st
     // set. The gate stays pending, saying it waits for the group, unless a composed verdict already stands on it.
     const out: CommitStatus[] = [];
     if (!(gateNow ?? "").includes(" composed: ")) {
-      out.push({ state: "pending", context: CHECK_CONTEXT, description: "waits for its group's composed check" });
+      out.push({ state: "pending", context: CHECK_CONTEXT, description: WAITS_FOR_COMPOSED });
     }
     const repo = c["repo-check"];
     if (repo) out.push(judgedFirst(c, repoStatus(repo, facts)));

modules/gitea/test/guard.test.ts

@@ -339,6 +339,30 @@ test("gitea_merge_pull_request refuses a verdict judged against an earlier trunk
   assert.equal(state.emitted.length, 4, "a merge asks no check");
 });
 
+// **A group member whose gate waits for its group's composed check is not announced again by the merge tool** (novox/hq issue
+// 1106): the member's own check can only say "composed" again, so each refusal's ask brought a verdict that names no trunk and
+// the next merge was refused the same way, for ever. The tool says whose verdict is awaited and asks nothing.
+test("gitea_merge_pull_request asks no check of a member that waits for its group's composed check", async () => {
+  const merge = () => tool("gitea_merge_pull_request").run({ owner: "novox", repo: "mesh-catalog", number: 7, head: REVIEWED });
+  state.trunk = TRUNK;
+  const before = state.emitted.length;
+  state.statuses = [{ context: "mesh/merge-gate", state: "pending", description: "waits for its group's composed check" },
+    { context: "mesh/repo-check", state: "success", description: "pass: y" }];
+  for (let i = 0; i < 3; i++) {
+    await assert.rejects(merge(), (err: Error) => {
+      assert.match(err.message, /waits for its group's composed check/);
+      assert.match(err.message, /mesh-delivery asks it, this tool asks no check/);
+      assert.match(err.message, /Nothing was merged/);
+      return true;
+    });
+  }
+  assert.equal(state.emitted.length, before, "no check is announced again and again");
+  assert.equal(state.merged, false);
+  // Once the composed verdict names the member's unmoved trunk, it merges.
+  state.statuses = judgedAt(TRUNK);
+  assert.equal(((await merge()) as Record<string, unknown>).merged, true);
+});
+
 // A merge lands only the head that was reviewed (novox/hq issue 381): mesh-controller#197 was reviewed at one head,
 // force-pushed, checked green at the new one, and merged there, because the tool merged whatever the head was.
 test("gitea_merge_pull_request merges only the head it is named, and refuses one that is not the head", async () => {

modules/gitea/trunk.ts

@@ -17,7 +17,7 @@
 import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
 import { dirname } from "node:path";
 import type { CommitStatus, GiteaClient, GiteaPull, HeadCheck } from "./client.js";
-import { CHECK_CONTEXT, REPO_CHECK_CONTEXT } from "./pulls.js";
+import { CHECK_CONTEXT, REPO_CHECK_CONTEXT, WAITS_FOR_COMPOSED } from "./pulls.js";
 
 /** A commit as a verdict names it: its first twelve characters. */
 export function twelve(sha: string | undefined): string {
@@ -58,6 +58,11 @@ export function checkOnItsWay(gate: HeadCheck | undefined): boolean {
   return d.startsWith("waiting in the build queue") || d.startsWith("building on ") || d.startsWith("checking again");
 }
 
+/** Whether a gate says it waits for its group's composed check. */
+export function waitsForComposed(gate: HeadCheck | undefined): boolean {
+  return !!gate && gate.state === "pending" && (gate.description ?? "").startsWith(WAITS_FOR_COMPOSED);
+}
+
 /**
  * The statuses a trunk's move puts back to pending on one open pull request's head, from what the forge holds on it
  * now: none when it has no merge check yet, when a check is pending (on its way, or already out of date — its

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go

@@ -634,6 +634,7 @@ func (h *Holder) answered(key string, ans sdk.Answer) []func() {
 	for _, d := range covered {
 		r := d.Ready
 		r.Warrant = ans.Words
+		r.Repeats = 0      // the operator chose: whatever the tool refused before is asked afresh (issue 1106)
 		h.mergeAgainNow(d) // the operator chose: the merge is asked at once
 		switch ans.Option {
 		case OptionApprove:

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1348_test.go

@@ -173,3 +173,43 @@ func TestAGroupMergedWholeMovesSourceMainOnceAndAnnouncesWithTheTeller(t *testin
 		t.Fatalf("source.main moved again: %d", len(w.source.moves))
 	}
 }
+
+// 1106: the same refusal again and again with the trunk not moved is not asked on without end. Three refusals in a row with
+// the same cause end the asking by a verdict, a status or a composed check, and say it urgent; a trunk's move, or the
+// operator's Approve, starts it again.
+func TestARefusalThatRepeatsWithNoTrunkMoveIsNotAskedAgainWithoutEnd(t *testing.T) {
+	w := answeredFirst(t)
+	d := w.h.deliveries[IDOf(gCat, gHk)]
+	refuse := func(took string) {
+		w.h.ForgeRefused(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
+			Words: "novox/mesh-catalog#2 is not merged: its verdict names no trunk commit it was judged against, and main is now f70a99dbf485. " +
+				"The check of head d2d2d2d2d2d2 merged into trunk f70a99dbf485 was asked; the last check took " + took + ". Ask again once both statuses pass"})
+	}
+	tried := w.now
+	for i, took := range []string{"4m1s", "4m9s"} {
+		refuse(took)
+		d.Ready.LastTry = tried
+		w.h.mergeAgainNow(d)
+		if !d.Ready.LastTry.IsZero() {
+			t.Fatalf("refusal %d: the merge was not asked again at once on a verdict, as after a first refusal", i+1)
+		}
+	}
+	refuse("4m5s") // the third in a row, with only the check's duration changed
+	if d.Ready.Repeats != 3 {
+		t.Fatalf("%d refusals counted for the same cause, wanted 3", d.Ready.Repeats)
+	}
+	d.Ready.LastTry = tried
+	w.h.mergeAgainNow(d) // the composed check or a verdict answers: nothing is asked
+	if d.Ready.LastTry.IsZero() {
+		t.Fatal("the merge was asked again after the same refusal three times with the trunk not moved")
+	}
+	if lines := urgentLines(w); len(lines) != 1 || !strings.Contains(lines[0].Says, "names no trunk commit") {
+		t.Fatalf("the repeated refusal is not said urgent: %+v", lines)
+	}
+	// The trunk moves: the merge is asked again.
+	d.TrunkNow = "a1a1a1a1a1a1a1a1"
+	w.h.mergeAgainNow(d)
+	if !d.Ready.LastTry.IsZero() {
+		t.Fatal("a trunk's move did not let the merge be asked again")
+	}
+}

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go

@@ -287,9 +287,20 @@ func (h *Holder) ForgeRefused(f sdk.Forge) {
 	if r.RefusedSince == nil {
 		r.RefusedSince = &now
 	}
+	r.noteRefusal(d, answer)
 	if !outOfDateRefusal(answer) {
 		r.Urgent = true // never a warning again for this head
 	}
+	if r.Repeats >= maxRepeatedRefusals {
+		// Loud, not endless (novox/hq issue 1106): the same refusal again with the trunk not moved is no trunk's move, so
+		// the merge is not asked again by a verdict or a status; the operator is told, urgent, and decides.
+		r.Urgent = true
+		if r.Repeats == maxRepeatedRefusals {
+			h.Logf("[mesh-delivery] %s: the merge tool refused it %d times in a row with the same words and the trunk not moved, so "+
+				"the merge is asked no more until the trunk moves, a new head is announced or the operator approves again: %s",
+				d.ID, r.Repeats, answer)
+		}
+	}
 	if answer != r.Answer {
 		h.Logf("[mesh-delivery] %s is approved, and the merge tool did not merge it (asked again at the next verdict, Approve or composed check): %s", d.ID,
 			answer)

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go

@@ -85,6 +85,11 @@ type ReadyRun struct {
 	// Urgent is set once a refusal of this head was one that is urgent at once: its severity never steps down to a
 	// warning after (issue 1275, AC6).
 	Urgent bool `json:"urgent,omitempty"`
+	// Repeats is how many times in a row the merge tool refused with the same words while the trunk stayed where RepeatTrunk
+	// says: past maxRepeatedRefusals the merge is asked no more by a verdict, a status or a composed check, and the refusal
+	// is urgent (novox/hq issue 1106). A trunk's move, a new head or the operator's Approve starts the count again.
+	Repeats     int    `json:"repeats,omitempty"`
+	RepeatTrunk string `json:"repeat_trunk,omitempty"`
 	// RefusedSince is when the merge tool began refusing it without a break.
 	RefusedSince *time.Time `json:"refused_since,omitempty"`
 	// Merged is when the merge tool merged it.
@@ -322,10 +327,43 @@ func (h *Holder) askFresh(d *Delivery, now time.Time) func() {
 // verdict, an Approve, a group's composed check), so the wait after a refusal does not hold it. Under the lock.
 func (h *Holder) mergeAgainNow(d *Delivery) {
 	if d != nil && d.Ready != nil && d.Ready.Merged == nil {
+		if d.Ready.repeatedRefusal(d) {
+			return // the same refusal again and again with the trunk where it was: asking more changes nothing (issue 1106)
+		}
 		d.Ready.LastTry = time.Time{}
 	}
 }
 
+// maxRepeatedRefusals is how many refusals in a row, with the same words and the trunk not moved, end the asking again: a
+// refusal that repeats with no trunk move is not a trunk's move, and the check asked after each only brings the same words.
+const maxRepeatedRefusals = 3
+
+// repeatedRefusal is whether the merge was refused maxRepeatedRefusals times with these words and the trunk where it is
+// now: it is then asked no more until something changes (a trunk's move, a new head, the operator's Approve).
+func (r *ReadyRun) repeatedRefusal(d *Delivery) bool {
+	return r.Repeats >= maxRepeatedRefusals && r.RepeatTrunk == d.TrunkNow
+}
+
+// refusalKey is a refusal's first sentence, up to the trunk it names ("... and main is now <t>"): what follows says how long the
+// last check took and the like, which differs each time and is no change of cause.
+func refusalKey(answer string) string {
+	if i := strings.Index(answer, " is now "); i >= 0 {
+		if j := strings.Index(answer[i:], ". "); j >= 0 {
+			return answer[:i+j]
+		}
+	}
+	return answer
+}
+
+// noteRefusal counts a refusal: the same words with the trunk where it was count on, anything else starts the count at one.
+func (r *ReadyRun) noteRefusal(d *Delivery, answer string) {
+	if refusalKey(answer) == refusalKey(r.Answer) && r.RepeatTrunk == d.TrunkNow && r.Repeats > 0 {
+		r.Repeats++
+		return
+	}
+	r.Repeats, r.RepeatTrunk = 1, d.TrunkNow
+}
+
 // readyWaits is what a delivery's merge waits on; empty when the Approve may be asked and the merge asked. A ready
 // delivery is asked. So is a proposed one whose verdict went out of date, when an Approve was asked for it already (it
 // stays through a trunk move, ADR 0421 §3) and its check against the trunk's head is not on its way; one without an

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client.go

@@ -119,9 +119,6 @@ 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.
@@ -150,8 +147,6 @@ 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
@@ -166,8 +161,6 @@ 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.
@@ -255,23 +248,6 @@ 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) {
@@ -290,7 +266,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{}, unplaced: map[string]bool{}, boundUsers: map[string]map[string]bool{}}
+		boundReady: 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
@@ -306,9 +282,6 @@ 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
@@ -353,10 +326,6 @@ 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
@@ -581,7 +550,6 @@ 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) {
@@ -692,7 +660,6 @@ 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) {
@@ -705,9 +672,6 @@ 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
 		}

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/client_test.go

@@ -200,14 +200,6 @@ 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-")
@@ -217,12 +209,6 @@ func newEngine(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)
@@ -247,7 +233,7 @@ func (e *engine) listen(t *testing.T) (stop func()) {
 			}()
 		}
 	}()
-	return func() { l.Close(); os.Remove(e.socket) }
+	return e
 }
 
 func (e *engine) at(path string) string { return filepath.Join(e.root, path) }
@@ -363,14 +349,10 @@ func aClient(t *testing.T, b *aBus, e *engine, state string) (*Client, func()) {
 		t.Fatal(err)
 	}
 	secrets.Within = 2 * time.Second
-	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,
+	c := &Client{Node: node, JS: js, Conn: nc, Secrets: secrets,
 		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
@@ -721,96 +703,3 @@ 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 := newEngine(t, map[string]string{dunstrc: "owned"})
-	stop := e.listen(t)
-	b.config(t, "dunst", "font-size", "13")
-	b.assign(t, "dunst", dunst(t))
-	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)
-
-	// 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)
-	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")
-}

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/main.go

@@ -133,11 +133,7 @@ func serve(ctx context.Context, credential, socket, stateDir string) error {
 	if err != nil {
 		return err
 	}
-	up, err := watchSocket(ctx, socket)
-	if err != nil {
-		return err
-	}
-	client := &Client{Node: c.Node, SocketUp: up, JS: js, Conn: nc, Secrets: secrets,
+	client := &Client{Node: c.Node, 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

modules/mesh-provisioner-client/cmd/mesh-provisioner-client/socketwatch.go

@@ -1,103 +0,0 @@
-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
-}

modules/mesh-provisioner-client/go.mod

@@ -6,7 +6,6 @@ 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 (
@@ -18,5 +17,6 @@ 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
 )