Wait for a held node without pinning a pool connection, and give up after a bounded wait naming it (hq ADR 0100)

cmd/mesh-controller/adopting_test.go

@@ -115,7 +115,7 @@ func TestTakingAModuleNotOnTheNodeIsRefused(t *testing.T) {
 
 func TestConvergingIsRefusedOnAPreviewThatWouldBeStale(t *testing.T) {
 	open, sent := anAdoptedAnchor(t)
-	_, err := converge(t.Context(), open, "anchor", false, "")
+	_, err := converge(t.Context(), open, "anchor", false, "", "")
 	if err == nil || !strings.Contains(err.Error(), "has not reported") {
 		t.Fatalf("a node that never reported was previewed: %v", err)
 	}
@@ -127,7 +127,7 @@ func TestConvergingIsRefusedOnAPreviewThatWouldBeStale(t *testing.T) {
 func TestTheFlipIsRefusedWhileAFoundContainerIsHeld(t *testing.T) {
 	open, sent := anAdoptedAnchor(t)
 	reportsHolding(t, open, heldContainer, heldFile)
-	_, err := converge(t.Context(), open, "anchor", true, "")
+	_, err := converge(t.Context(), open, "anchor", true, "", "")
 	if err == nil || !strings.Contains(err.Error(), "take anchor hello-web once its data has moved") {
 		t.Fatalf("the flip was not refused while hello-web holds its found container: %v", err)
 	}
@@ -157,7 +157,7 @@ func TestConvergingPreviewsThenChangesAndAdoptingKeepsWhatWasTaken(t *testing.T)
 	}
 	reportsHolding(t, open, heldFile)
 
-	preview, err := converge(ctx, open, "anchor", false, "")
+	preview, err := converge(ctx, open, "anchor", false, "", "")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -168,7 +168,7 @@ func TestConvergingPreviewsThenChangesAndAdoptingKeepsWhatWasTaken(t *testing.T)
 		// The anchor faces inward and is on the private network: the derived filter admits ssh
 		// from the mesh only.
 		"WILL CLOSE to everything outside the private network — ssh stays open from the mesh",
-		"notes\n      replacing the found file /etc/notes.conf (original kept at",
+		"notes\n      replacing the found file /etc/notes.conf (notes.conf), original kept at",
 		"assigns nftables",
 		"the found firewall (ufw) is disabled, never flushed",
 		// What it routes is not a listener: said not to be previewed, and to be dropped.
@@ -191,7 +191,7 @@ func TestConvergingPreviewsThenChangesAndAdoptingKeepsWhatWasTaken(t *testing.T)
 		t.Fatal("the preview changed something")
 	}
 
-	if _, err := converge(ctx, open, "anchor", true, ""); err != nil {
+	if _, err := converge(ctx, open, "anchor", true, digestIn(t, preview), ""); err != nil {
 		t.Fatal(err)
 	}
 	n, _ := open.inventory.NodeByName(ctx, "anchor")
@@ -245,7 +245,7 @@ func hasID(resources []map[string]any, id string) bool {
 func TestTheApiRefusesTheFlipInTheCommandLinesWords(t *testing.T) {
 	open, _ := anAdoptedAnchor(t)
 	reportsHolding(t, open, heldContainer)
-	_, direct := converge(t.Context(), open, "anchor", true, "")
+	_, direct := converge(t.Context(), open, "anchor", true, "", "")
 	if direct == nil {
 		t.Fatal("the flip was not refused")
 	}
@@ -283,7 +283,7 @@ func TestThePreviewSaysWhatNarrowsToTheMeshCloses(t *testing.T) {
 		{Protocol: "tcp", Address: "0.0.0.0", Port: 8080, By: "hello-web", Published: true,
 			ContainerPort: 80},
 	})
-	preview, err := converge(ctx, open, "anchor", false, "")
+	preview, err := converge(ctx, open, "anchor", false, "", "")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -307,3 +307,169 @@ func TestThePreviewSaysWhatNarrowsToTheMeshCloses(t *testing.T) {
 		t.Errorf("a port open to everywhere is not said to stay:\n%s", preview)
 	}
 }
+
+// digestIn is the digest a converge preview printed.
+func digestIn(t *testing.T, preview string) string {
+	t.Helper()
+	for _, line := range strings.Split(preview, "\n") {
+		if fields := strings.Fields(line); len(fields) == 2 && fields[0] == "preview" {
+			return fields[1]
+		}
+	}
+	t.Fatalf("the preview printed no digest:\n%s", preview)
+	return ""
+}
+
+// The flip acts on the preview the operator saw: it names that preview's digest, and it is refused
+// when the digest is missing, when anything the preview says has changed since, or when the node's
+// account of itself is too old to be the machine as it is.
+func TestTheFlipActsOnlyOnThePreviewTheOperatorSaw(t *testing.T) {
+	open, sent := anAdoptedAnchor(t)
+	ctx := t.Context()
+	if _, err := take(ctx, open, "anchor", "hello-web"); err != nil {
+		t.Fatal(err)
+	}
+	reportsHolding(t, open, heldFile)
+	preview, err := converge(ctx, open, "anchor", false, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	saw := digestIn(t, preview)
+	if !strings.Contains(preview, "converge anchor --yes "+saw) {
+		t.Fatalf("the preview does not say how to act on it:\n%s", preview)
+	}
+	unchanged := func() {
+		t.Helper()
+		if n, _ := open.inventory.NodeByName(ctx, "anchor"); !n.Adopted || len(*sent) != 0 {
+			t.Fatal("a refused flip changed something")
+		}
+	}
+
+	if _, err := converge(ctx, open, "anchor", true, "", ""); err == nil ||
+		!strings.Contains(err.Error(), "--yes "+saw) {
+		t.Fatalf("a flip naming no preview was not refused: %v", err)
+	}
+	unchanged()
+
+	// Something new is reachable: the preview the operator saw is not what would happen.
+	reportsReaching(t, open, []link.Reach{
+		{Protocol: "tcp", Address: "0.0.0.0", Port: 22, By: "sshd"},
+		{Protocol: "tcp", Address: "0.0.0.0", Port: 8080, By: "hello-web", Published: true,
+			ContainerPort: 80},
+		{Protocol: "tcp", Address: "0.0.0.0", Port: 6000, By: "something-new"},
+	}, heldFile)
+	_, err = converge(ctx, open, "anchor", true, saw, "")
+	if err == nil || !strings.Contains(err.Error(), "has changed since preview "+saw) {
+		t.Fatalf("a flip on a changed preview was not refused: %v", err)
+	}
+	unchanged()
+
+	// An account older than the flip trusts is refused, whatever digest is named.
+	again, err := converge(ctx, open, "anchor", false, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	saved := reportFreshFor
+	reportFreshFor = time.Nanosecond
+	_, err = converge(ctx, open, "anchor", true, digestIn(t, again), "")
+	reportFreshFor = saved
+	if err == nil || !strings.Contains(err.Error(), "wait for its next report") {
+		t.Fatalf("a flip on an old account was not refused: %v", err)
+	}
+	unchanged()
+
+	if _, err := converge(ctx, open, "anchor", true, digestIn(t, again), ""); err != nil {
+		t.Fatalf("the flip on the preview just seen was refused: %v", err)
+	}
+	if n, _ := open.inventory.NodeByName(ctx, "anchor"); n.Adopted {
+		t.Fatal("the flip did not converge the node")
+	}
+}
+
+// The flip holds the node from its checks to its send, so a push composed meanwhile waits and is
+// composed after it — never sent after it with the node still adopted. And the send inside the
+// flip is not made to wait on the flip's own hold.
+func TestTheFlipHoldsTheNodeWhileItSends(t *testing.T) {
+	open, _ := anAdoptedAnchor(t)
+	ctx := t.Context()
+	if _, err := take(ctx, open, "anchor", "hello-web"); err != nil {
+		t.Fatal(err)
+	}
+	reportsHolding(t, open, heldFile)
+	preview, err := converge(ctx, open, "anchor", false, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var heldElsewhere, heldHere error
+	sendNodes = func(inner context.Context, open *stores, names []string) error {
+		// Another caller cannot hold the node while the flip sends it.
+		waiting, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
+		defer cancel()
+		if release, err := open.inventory.HoldNodes(waiting, names); err == nil {
+			release()
+			heldElsewhere = errors.New("another caller held the node while the flip sent it")
+		}
+		// The flip's own send holds it without waiting on itself.
+		_, release, err := holdNodes(inner, open, names)
+		if err != nil {
+			heldHere = err
+			return err
+		}
+		release()
+		return nil
+	}
+	if _, err := converge(ctx, open, "anchor", true, digestIn(t, preview), ""); err != nil {
+		t.Fatal(err)
+	}
+	if heldElsewhere != nil || heldHere != nil {
+		t.Fatalf("%v %v", heldElsewhere, heldHere)
+	}
+	// And given back once it is done.
+	release, err := open.inventory.HoldNodes(ctx, []string{"anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	release()
+}
+
+// novox/hq ADR 0103: the preview names every kind of thing a module the flip takes holds as found,
+// not only its files, and the digest changes when any of them does.
+func TestThePreviewNamesEveryHeldKind(t *testing.T) {
+	open, _ := anAdoptedAnchor(t)
+	ctx := t.Context()
+	if _, err := take(ctx, open, "anchor", "hello-web"); err != nil {
+		t.Fatal(err)
+	}
+	since := time.Now()
+	held := []link.Held{heldFile,
+		{ID: "notes.data", Module: "notes", Kind: "directory", Target: "/var/lib/notes", Since: since},
+		{ID: "notes.daemon", Module: "notes", Kind: "service", Target: "notes.service", Since: since},
+		{ID: "notes.seed", Module: "notes", Kind: "archive", Target: "/srv/notes", Since: since},
+		{ID: "notes.worker", Module: "notes", Kind: "process", Target: "notes-worker", Since: since},
+		{ID: "notes.account", Module: "notes", Kind: "user", Target: "notes", Since: since},
+	}
+	reportsHolding(t, open, held...)
+	preview, err := converge(ctx, open, "anchor", false, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{
+		"replacing the found directory /var/lib/notes (notes.data)",
+		"replacing the found service notes.service (notes.daemon)",
+		"replacing the found archive /srv/notes (notes.seed)",
+		"replacing the found process notes-worker (notes.worker)",
+		"replacing the found user notes (notes.account)",
+	} {
+		if !strings.Contains(preview, want) {
+			t.Errorf("the preview does not say %q:\n%s", want, preview)
+		}
+	}
+	reportsHolding(t, open, held[:len(held)-1]...)
+	fewer, err := converge(ctx, open, "anchor", false, "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if digestIn(t, fewer) == digestIn(t, preview) {
+		t.Fatal("the digest does not change with what is held")
+	}
+}

cmd/mesh-controller/adoption.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"flag"
 	"fmt"
@@ -117,7 +119,7 @@ func take(ctx context.Context, open *stores, node, module string) (string, error
 	var replaces []string
 	for _, h := range reported.Held {
 		if h.Module == module {
-			replaces = append(replaces, fmt.Sprintf("  %s %s (%s)", h.Kind, h.Target, h.ID))
+			replaces = append(replaces, "  "+heldLine(h))
 		}
 	}
 	if len(replaces) > 0 {
@@ -127,6 +129,10 @@ func take(ctx context.Context, open *stores, node, module string) (string, error
 	return said + fmt.Sprintf("\n  run `push %s` to cut it over", node), nil
 }
 
+// reportFreshFor is how old a node's account of itself may be for the flip to act on it. A
+// variable so a test can age a report without waiting.
+var reportFreshFor = 15 * time.Minute
+
 // converge previews, and with yes makes, the flip of an adopted node to converged: every module it
 // runs is taken, the filter module is assigned to load the mesh's derived filter in place of the
 // guard, and the found firewall is retired — disabled, never flushed — by the host.
@@ -134,11 +140,28 @@ func take(ctx context.Context, open *stores, node, module string) (string, error
 // Refused while an assigned module still holds a found container: each service is taken on its
 // own, when its data has moved, never by the flip. And refused on a preview that would be stale:
 // what is reachable is the node's last account, so that account must be of what it was last sent.
-func converge(ctx context.Context, open *stores, node string, yes bool, filter string) (string, error) {
+//
+// **The flip acts on the preview the operator saw** and on nothing else. The preview ends with a
+// short digest of what it said — every reachable thing and its fate, the modules the flip takes and
+// the filter — and yes must name that digest: if anything the preview would say has changed since,
+// the flip is refused rather than done on a preview nobody read. And it is refused on an account
+// older than reportFreshFor: what was reachable then is not evidence of what is reachable now.
+func converge(ctx context.Context, open *stores, node string, yes bool, digest string,
+	filter string) (string, error) {
 	inv := open.inventory
 	if filter == "" {
 		filter = DefaultFilter
 	}
+	if yes {
+		// Held from the checks to the send, so no push composed before the flip is sent after it
+		// and returns the node to adopted.
+		held, release, err := holdNodes(ctx, open, []string{node})
+		if err != nil {
+			return "", err
+		}
+		defer release()
+		ctx = held
+	}
 	record, err := inv.NodeByName(ctx, node)
 	if err != nil {
 		return "", err
@@ -219,9 +242,25 @@ func converge(ctx context.Context, open *stores, node string, yes bool, filter s
 	}
 	derived := derivedFilter{rules: rules, foundation: with.Foundation, mesh: with.Mesh,
 		outward: plan.PublicDomain != ""}
-	preview := previewOf(node, reported, derived, plan, taken, filter, runs[filter])
+	preview, saw := previewOf(node, reported, derived, plan, taken, filter, runs[filter])
+	preview += "\n\n  preview " + saw
 	if !yes {
-		return preview + fmt.Sprintf("\n\nNothing has changed. Run `converge %s --yes` to do it.", node), nil
+		return preview + fmt.Sprintf("\n\nNothing has changed. Run `converge %s --yes %s` to do "+
+			"it.", node, saw), nil
+	}
+	if age := time.Since(reported.At); age > reportFreshFor {
+		return preview, fmt.Errorf("%s last said what is reachable on it %s ago, and the flip acts "+
+			"only on an account newer than %s: wait for its next report, or run `push %s --wait 2m`, "+
+			"then preview again", node, age.Round(time.Second), reportFreshFor, node)
+	}
+	if digest == "" {
+		return preview, fmt.Errorf("converging %s acts on the preview you saw: name its digest, "+
+			"`converge %s --yes %s`, once you have read it", node, node, saw)
+	}
+	if digest != saw {
+		return preview, fmt.Errorf("what converging %s would do has changed since preview %s "+
+			"(it is now %s): read the preview above, and run `converge %s --yes %s` if it is "+
+			"what you want", node, digest, saw, node, saw)
 	}
 
 	// The flip. The filter first, and only kept if the node still resolves with it: a node that
@@ -253,9 +292,12 @@ func converge(ctx context.Context, open *stores, node string, yes bool, filter s
 	return said + "\n  sent: the host loads the mesh's filter and disables the firewall it found", nil
 }
 
-// previewOf is what converging a node will change, before it changes it.
+// previewOf is what converging a node will change, before it changes it, and a short digest of
+// what it said: every reachable thing and its fate, the modules the flip takes and the filter. The
+// digest is what the flip is asked to act on, so it changes whenever any of those would.
 func previewOf(node string, reported inventory.Adoption, derived derivedFilter,
-	plan catalogue.Resolution, taken []string, filter string, filterAssigned bool) string {
+	plan catalogue.Resolution, taken []string, filter string, filterAssigned bool) (string, string) {
+	var said []string
 	var b strings.Builder
 	fmt.Fprintf(&b, "converging %s\n", node)
 	fmt.Fprintf(&b, "\n  reachable on the machine now, as it reported at %s:\n",
@@ -271,7 +313,9 @@ func previewOf(node string, reported inventory.Adoption, derived derivedFilter,
 		if r.Published {
 			what += fmt.Sprintf(" (published, container port %d)", r.ContainerPort)
 		}
-		fmt.Fprintf(&b, "    %-44s %s\n", what, derived.fate(r))
+		fate := derived.fate(r)
+		fmt.Fprintf(&b, "    %-44s %s\n", what, fate)
+		said = append(said, fmt.Sprintf("reach %s %s %s", r.Address, what, fate))
 	}
 	if len(reported.Reachable) == 0 {
 		b.WriteString("    nothing reported\n")
@@ -302,14 +346,19 @@ func previewOf(node string, reported inventory.Adoption, derived derivedFilter,
 	}
 	for _, m := range takes {
 		fmt.Fprintf(&b, "    %s\n", m)
+		said = append(said, "take "+m)
+		// Every kind it holds — a directory, a service, an archive, a process, a user as well as a
+		// file (novox/hq ADR 0103) — each said, and each part of what the flip is asked to act on.
 		for _, h := range reported.Held {
-			if h.Module == m && h.Kind == "file" {
-				fmt.Fprintf(&b, "      replacing the found file %s", h.Target)
-				if h.Kept != "" {
-					fmt.Fprintf(&b, " (original kept at %s)", h.Kept)
-				}
-				b.WriteString("\n")
+			if h.Module != m {
+				continue
 			}
+			fmt.Fprintf(&b, "      replacing the found %s", heldLine(h))
+			if h.Kept != "" {
+				fmt.Fprintf(&b, ", original kept at %s", h.Kept)
+			}
+			b.WriteString("\n")
+			said = append(said, "replace "+m+" "+heldLine(h)+" "+h.Kept)
 		}
 	}
 	if !filterAssigned {
@@ -321,7 +370,11 @@ func previewOf(node string, reported inventory.Adoption, derived derivedFilter,
 	} else {
 		fmt.Fprintf(&b, "  the found firewall (%s) is disabled, never flushed: its configuration stays on disk\n", fw)
 	}
-	return strings.TrimRight(b.String(), "\n")
+	said = append(said, fmt.Sprintf("filter %s assigned=%t firewall=%s", filter, filterAssigned, fw))
+	// Sorted: the same account, reported in another order, is the same preview.
+	sort.Strings(said)
+	sum := sha256.Sum256([]byte(strings.Join(said, "\n")))
+	return strings.TrimRight(b.String(), "\n"), hex.EncodeToString(sum[:])[:12]
 }
 
 // derivedFilter is what the filter the flip loads is rendered from, as AsNftables renders it.
@@ -378,6 +431,11 @@ func (d derivedFilter) fate(r inventory.Reach) string {
 	return "WILL CLOSE — no module assigned here declares it"
 }
 
+// heldLine is one thing a node holds as found, as take and the converge preview both say it.
+func heldLine(h inventory.Held) string {
+	return fmt.Sprintf("%s %s (%s)", h.Kind, h.Target, h.ID)
+}
+
 // loopback is an address nothing off the machine reaches.
 func loopback(address string) bool {
 	a := strings.Trim(address, "[]")
@@ -396,6 +454,12 @@ func adopt(ctx context.Context, open *stores, node string) (string, error) {
 	if record.Adopted {
 		return "", fmt.Errorf("%s is adopted already", node)
 	}
+	held, release, err := holdNodes(ctx, open, []string{node})
+	if err != nil {
+		return "", err
+	}
+	defer release()
+	ctx = held
 	if err := inv.SetAdopted(ctx, node, true); err != nil {
 		return "", err
 	}
@@ -416,17 +480,18 @@ func takeCommand(ctx context.Context, args []string) error {
 
 func convergeCommand(ctx context.Context, args []string) error {
 	set := flag.NewFlagSet("converge", flag.ContinueOnError)
-	yes := set.Bool("yes", false, "do it; without it, only the preview")
+	yes := set.String("yes", "", "do it, naming the digest the preview printed; without it, only "+
+		"the preview")
 	filter := set.String("filter", DefaultFilter, "the module that loads the mesh's filter")
 	positionals, err := parseAround(set, args)
 	if err != nil {
 		return err
 	}
 	if len(positionals) != 1 {
-		return errors.New("converge <node> [--yes] [--filter nftables]")
+		return errors.New("converge <node> [--yes <digest>] [--filter nftables]")
 	}
 	return runAct(ctx, func(open *stores) (string, error) {
-		return converge(ctx, open, positionals[0], *yes, *filter)
+		return converge(ctx, open, positionals[0], *yes != "", *yes, *filter)
 	})
 }
 

cmd/mesh-controller/api.go

@@ -105,7 +105,7 @@ func commands(who Authenticator) http.Handler {
 		return take(ctx, open, in.Node, in.Module)
 	}))
 	mux.HandleFunc("POST /converge", acting(who, false, func(ctx context.Context, open *stores, in request) (string, error) {
-		return converge(ctx, open, in.Node, in.Yes, in.Filter)
+		return converge(ctx, open, in.Node, in.Yes, in.Digest, in.Filter)
 	}))
 	mux.HandleFunc("POST /adopt", acting(who, false, func(ctx context.Context, open *stores, in request) (string, error) {
 		return adopt(ctx, open, in.Node)
@@ -124,9 +124,10 @@ func commands(who Authenticator) http.Handler {
 type request struct {
 	Node   string `json:"node"`
 	Module string `json:"module"`
-	// Yes and Filter are converge's: do it rather than preview it, and which module loads the
-	// mesh's filter.
+	// Yes, Digest and Filter are converge's: do it rather than preview it, the digest of the
+	// preview it acts on, and which module loads the mesh's filter.
 	Yes    bool   `json:"yes,omitempty"`
+	Digest string `json:"digest,omitempty"`
 	Filter string `json:"filter,omitempty"`
 }
 

cmd/mesh-controller/hold.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"context"
+)
+
+// heldKey carries the nodes this call already holds, so an act that holds a node and then sends
+// through sendTo does not wait on itself.
+type heldKey struct{}
+
+// holdNodes holds the named nodes for composing and sending their declarations (novox/hq ADR
+// 0100), skipping any the context already holds, and returns a context that says it holds them.
+// Release gives back only what this call took.
+func holdNodes(ctx context.Context, open *stores, names []string) (context.Context, func(), error) {
+	already, _ := ctx.Value(heldKey{}).(map[string]bool)
+	var take []string
+	for _, n := range names {
+		if !already[n] {
+			take = append(take, n)
+		}
+	}
+	if len(take) == 0 {
+		return ctx, func() {}, nil
+	}
+	release, err := open.inventory.HoldNodes(ctx, take)
+	if err != nil {
+		return ctx, nil, err
+	}
+	held := map[string]bool{}
+	for n := range already {
+		held[n] = true
+	}
+	for _, n := range take {
+		held[n] = true
+	}
+	return context.WithValue(ctx, heldKey{}, held), release, nil
+}

cmd/mesh-controller/hold_test.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+// A cascade round gives its hold back on every way out: a declaration that cannot be marshalled
+// and a send that fails leave nobody waiting for the node.
+func TestASendRoundGivesItsHoldBackOnEveryWayOut(t *testing.T) {
+	open := aMesh(t)
+	ctx := t.Context()
+	unmarshallable := func(context.Context, string) (sendable, error) {
+		return sendable{Resources: []map[string]any{{"id": "x", "bad": make(chan int)}}}, nil
+	}
+	plain := func(context.Context, string) (sendable, error) {
+		return sendable{Resources: []map[string]any{{"id": "x"}}}, nil
+	}
+	failing := func(readyNode, []byte) error { return errors.New("the broker went away") }
+	fine := func(readyNode, []byte) error { return nil }
+
+	for name, round := range map[string]func() error{
+		"a body that cannot be marshalled": func() error {
+			_, err := sendRound(ctx, open, []string{"anchor"}, unmarshallable, fine)
+			return err
+		},
+		"a send that fails": func() error {
+			_, err := sendRound(ctx, open, []string{"anchor"}, plain, failing)
+			return err
+		},
+	} {
+		if err := round(); err == nil {
+			t.Fatalf("%s was not an error", name)
+		}
+		waiting, cancel := context.WithTimeout(ctx, 2*time.Second)
+		release, err := open.inventory.HoldNodes(waiting, []string{"anchor"})
+		cancel()
+		if err != nil {
+			t.Fatalf("after %s the node is still held: %v", name, err)
+		}
+		release()
+	}
+}

cmd/mesh-controller/main.go

@@ -162,7 +162,7 @@ func usage() {
   assign <node> <module>         put a module on a node
   unassign <node> <module>       take it off
   take <node> <module>           cut a module over on an adopted node, once its data has moved
-  converge <node> [--yes] [--filter nftables]   preview, then make, an adopted node converged
+  converge <node> [--yes <digest>] [--filter nftables]   preview, then make, an adopted node converged
   adopt <node>                   return a converged node to adopted; what was taken stays taken
   settings set <module> <file>   what a module's config should say, for the whole mesh
   settings set <module> <file> --node <n>   ...or for one machine

cmd/mesh-controller/nodes.go

@@ -279,13 +279,20 @@ func issueFor(ctx context.Context, inv *inventory.Inventory, existing, fresh str
 		}
 		name = node.Name
 	}
-	// Saying adopted makes the node adopted. Not saying it leaves the node as it is: re-issuing a
-	// token for an adopted node does not converge it — converging is its own act, previewed
-	// (novox/hq ADR 0100).
-	if adopted {
-		if err := inv.SetAdopted(ctx, name, true); err != nil {
+	// Not saying adopted leaves the node as it is: re-issuing a token for an adopted node does not
+	// converge it — converging is its own act, previewed (novox/hq ADR 0100). And saying it for a
+	// node already converged is refused rather than done quietly: returning a node to adopted is
+	// its own act too, which unloads the mesh's filter and enables the found firewall again.
+	if adopted && fresh == "" {
+		node, err := inv.NodeByName(ctx, name)
+		if err != nil {
 			return inventory.Issued{}, err
 		}
+		if !node.Adopted {
+			return inventory.Issued{}, fmt.Errorf("%s is converged, and a token does not change "+
+				"that: run `adopt %s` to return it to adopted, then issue the token without "+
+				"--adopted", name, name)
+		}
 	}
 
 	return inv.IssueToken(ctx, name, validFor)

cmd/mesh-controller/nodes_test.go

@@ -139,11 +139,16 @@ func TestATokenIssuedAdoptedSaysSoAndReissuingDoesNotConverge(t *testing.T) {
 	if !again.Node.Adopted {
 		t.Fatal("re-issuing without --adopted converged the node; converging is its own act")
 	}
-	existing, err := issueFor(ctx, open.inventory, "laptop", "", true, time.Hour)
-	if err != nil {
+	// --adopted for a node already converged is refused, and points at the act that does it.
+	if _, err := issueFor(ctx, open.inventory, "laptop", "", true, time.Hour); err == nil ||
+		!strings.Contains(err.Error(), "adopt laptop") {
+		t.Fatalf("--adopted on a converged node was not refused: %v", err)
+	}
+	if n, _ := open.inventory.NodeByName(ctx, "laptop"); n.Adopted {
+		t.Fatal("a refused token flipped the node to adopted")
+	}
+	// And said for a node that is adopted already, it is the ordinary re-issue.
+	if _, err := issueFor(ctx, open.inventory, "joiner", "", true, time.Hour); err != nil {
 		t.Fatal(err)
 	}
-	if !existing.Node.Adopted {
-		t.Fatal("--adopted on an existing record did not make it adopted")
-	}
 }

cmd/mesh-controller/plan.go

@@ -384,6 +384,20 @@ func renderingFor(ctx context.Context, open *stores, node string,
 			given[m.Module] = g
 		}
 	}
+	// And one holder per machine port across the node's modules: settings written before this was
+	// refused where they are set are refused here rather than composed into two containers on
+	// one port.
+	holders := map[int]string{}
+	for _, m := range plan.Modules {
+		for _, at := range given[m.Module] {
+			if other, twice := holders[at]; twice && other != m.Module {
+				return catalogue.Rendering{}, inventory.Node{}, fmt.Errorf("%w: %s and %s are "+
+					"both given machine port %d on %s", inventory.ErrPortTaken, other, m.Module,
+					at, node)
+			}
+			holders[at] = m.Module
+		}
+	}
 
 	ports := map[string]map[int]int{}
 	for _, m := range plan.Modules {

cmd/mesh-controller/push.go

@@ -282,8 +282,14 @@ func pushCommand(ctx context.Context, args []string) error {
 		asked = append(asked, n.Name)
 	}
 
+	// Held from composing to sending, so a converge on one of them cannot send between the two
+	// and be overtaken by what was composed before it (novox/hq ADR 0100).
+	held, release, err := holdNodes(ctx, open, asked)
+	if err != nil {
+		return err
+	}
 	sending, refusals := composeEach(asked, func(node string) (sendable, error) {
-		plan, settings, err := planFor(ctx, open, node)
+		plan, settings, err := planFor(held, open, node)
 		if err != nil {
 			return sendable{}, err
 		}
@@ -294,10 +300,11 @@ func pushCommand(ctx context.Context, args []string) error {
 		// The private network is in here with everything else. It used to be composed separately
 		// and prepended, which meant every machine with an address was on it and no machine could
 		// be kept off. It is a module now, so it arrives the way a module does.
-		return declarationWith(ctx, open, node, plan, settings, gens, Allocating)
+		return declarationWith(held, open, node, plan, settings, gens, Allocating)
 	})
 
 	sentDigest := map[string]string{}
+	defer release()
 	for _, s := range sending {
 		body, err := s.declared.Body()
 		if err != nil {
@@ -319,6 +326,7 @@ func pushCommand(ctx context.Context, args []string) error {
 		sentDigest[s.node] = digest
 		fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
 	}
+	release()
 	fmt.Printf("\n%d node(s) told\n", len(sending))
 
 	// **A named push leaves the mesh consistent, not just the machine it named** (novox/hq
@@ -373,31 +381,35 @@ func pushCommand(ctx context.Context, args []string) error {
 			// (novox/hq ADR 0066). The earlier cut routed these through sendTo, which is
 			// all-or-nothing — so one swept machine's compose error failed the operator's named
 			// push and skipped its --wait, the very intolerance the main path exists to avoid.
-			sending, refused := composeEach(also, func(node string) (sendable, error) {
-				plan, settings, err := planFor(ctx, open, node)
-				if err != nil {
-					return sendable{}, err
-				}
-				reportUnhostable(node, plan)
-				return declarationWith(ctx, open, node, plan, settings, gens, Allocating)
-			})
+			// Held for this round only, and after the last round's were given back, so two pushes
+			// cascading into each other's machines never each wait on the other.
+			refused, err := sendRound(ctx, open, also,
+				func(held context.Context, node string) (sendable, error) {
+					plan, settings, err := planFor(held, open, node)
+					if err != nil {
+						return sendable{}, err
+					}
+					reportUnhostable(node, plan)
+					return declarationWith(held, open, node, plan, settings, gens, Allocating)
+				},
+				func(s readyNode, body []byte) error {
+					if err := link.Declare(ctx, server.Channel(), ident, s.node, body,
+						15*time.Second); err != nil {
+						return err
+					}
+					record, err := inv.NodeByName(ctx, s.node)
+					if err != nil {
+						return err
+					}
+					if err := inv.RecordSent(ctx, record.ID, digestOf(body)); err != nil {
+						return err
+					}
+					fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
+					return nil
+				})
 			refusals = append(refusals, refused...)
-			for _, s := range sending {
-				body, err := s.declared.Body()
-				if err != nil {
-					return err
-				}
-				if err := link.Declare(ctx, server.Channel(), ident, s.node, body, 15*time.Second); err != nil {
-					return err
-				}
-				record, err := inv.NodeByName(ctx, s.node)
-				if err != nil {
-					return err
-				}
-				if err := inv.RecordSent(ctx, record.ID, digestOf(body)); err != nil {
-					return err
-				}
-				fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.declared.Resources))
+			if err != nil {
+				return err
 			}
 			// Every candidate this round is marked handled — the sent ones so they are not
 			// re-listed, and the refused ones so a machine that cannot be composed does not make
@@ -498,6 +510,33 @@ func composeEach(names []string,
 	return sending, refusals
 }
 
+// sendRound holds the named nodes, composes each and sends each that composed, and gives the hold
+// back on every way out — a body that cannot be marshalled and a send that fails included
+// (novox/hq ADR 0100). A node that cannot be composed is a refusal, not an error: the others are
+// still sent.
+func sendRound(ctx context.Context, open *stores, names []string,
+	compose func(held context.Context, node string) (sendable, error),
+	send func(s readyNode, body []byte) error) ([]string, error) {
+	held, release, err := holdNodes(ctx, open, names)
+	if err != nil {
+		return nil, err
+	}
+	defer release()
+	sending, refused := composeEach(names, func(node string) (sendable, error) {
+		return compose(held, node)
+	})
+	for _, s := range sending {
+		body, err := s.declared.Body()
+		if err != nil {
+			return refused, err
+		}
+		if err := send(s, body); err != nil {
+			return refused, err
+		}
+	}
+	return refused, nil
+}
+
 // couldNotBeResolved is what a push ends with when some machines could not be worked out.
 //
 // **After the rest have been sent, never instead of sending them.** It is still an error, because
@@ -532,6 +571,14 @@ func sendTo(ctx context.Context, open *stores, names []string) error {
 		return err
 	}
 
+	// Held from composing to sending (novox/hq ADR 0100); a caller that holds them already —
+	// converge, which flips the node and then sends it — is not made to wait on itself.
+	ctx, release, err := holdNodes(ctx, open, names)
+	if err != nil {
+		return err
+	}
+	defer release()
+
 	var sending []readyNode
 	var refusals []string
 	for _, name := range names {

cmd/mesh-controller/sendable.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"sort"
+	"strings"
 
 	"github.com/novox/mesh-controller/internal/catalogue"
 	"github.com/novox/mesh-controller/internal/inventory"
@@ -23,8 +24,8 @@ type sendable struct {
 }
 
 // adoptionEnvelope is what an adopted node is told about its mode. Taken is every module taken on
-// it that it runs; Untaken is, for every module it runs that is not taken, the ids of that module's
-// files and containers — the resources the host keeps as found until the module is taken. Ids
+// it that it runs; Untaken is, for every module it runs that is not taken, the ids of every one of
+// that module's resources — what the host keeps as found until the module is taken (ADR 0103). Ids
 // rather than a rule to split them by, because a module's name may contain a dot.
 type adoptionEnvelope struct {
 	Taken   []string            `json:"taken"`
@@ -75,12 +76,11 @@ func adoptionFor(plan catalogue.Resolution, taken []string,
 	for _, r := range composed.Resources {
 		id, _ := r["id"].(string)
 		module, owned := composed.Owner[id]
-		if !owned || !runs[module] || isTaken[module] {
-			continue
-		}
-		switch r["type"] {
-		case "file", "container":
-		default:
+		// Every kind, not only files and containers (novox/hq ADR 0103): a directory, a service, a
+		// container mounting what was found and an action run in a held container all reach what
+		// the machine already has. What the mesh declares of its own is never held.
+		if !owned || !runs[module] || isTaken[module] ||
+			strings.HasPrefix(id, catalogue.AdoptionPrefix) {
 			continue
 		}
 		if out.Untaken == nil {

cmd/mesh-controller/sendable_test.go

@@ -84,8 +84,10 @@ func TestAnAdoptedDeclarationCarriesItsModeAndWhatWasTaken(t *testing.T) {
 		t.Fatal("an adopted node's declaration does not say it is adopted")
 	}
 	untaken := declared.Adoption.Untaken["hello-web"]
-	if !reflect.DeepEqual(untaken, []string{"hello-web.page", "hello-web.server"}) {
-		t.Fatalf("hello-web's files and containers are not named untaken: %v", declared.Adoption)
+	// Every kind — its directory too (novox/hq ADR 0103).
+	if !reflect.DeepEqual(untaken, []string{"hello-web.page", "hello-web.server",
+		"hello-web.served"}) {
+		t.Fatalf("hello-web's resources are not all named untaken: %v", declared.Adoption)
 	}
 	if len(declared.Adoption.Taken) != 0 {
 		t.Fatalf("nothing was taken, and the declaration says %v", declared.Adoption.Taken)

internal/catalogue/adoption.go

@@ -207,8 +207,10 @@ func GuardUnitText() string {
 }
 
 // GuardResources are the guard as four resources of the existing kinds: the tool that loads it,
-// the table, the unit, and the unit running, restarted when the table changes. Nothing when there
-// is nothing to guard: an empty set is not a table nft loads.
+// the table, the unit, and the unit running — reloaded when the table changes, so the new table
+// replaces the old in one `nft -f` through the unit's ExecReload with no moment unguarded, and
+// restarted only when the unit itself changes. Nothing when there is nothing to guard: an empty
+// set is not a table nft loads.
 func GuardResources(ports []int) []map[string]any {
 	if len(ports) == 0 {
 		return nil
@@ -220,6 +222,6 @@ func GuardResources(ports []int) []map[string]any {
 		{"id": GuardUnitID(), "type": "file", "path": GuardUnitPath, "content": GuardUnitText(),
 			"mode": "0644"},
 		{"id": GuardRunningID(), "type": "service", "unit": GuardUnit, "state": "running",
-			"boot": "enabled", "restart-on": []any{GuardID(), GuardUnitID()}},
+			"boot": "enabled", "restart-on": []any{GuardUnitID()}, "reload-on": []any{GuardID()}},
 	}
 }

internal/catalogue/adoption_test.go

@@ -174,8 +174,12 @@ func TestAnAdoptedNodeLoadsNoFilterOfTheMeshs(t *testing.T) {
 	if pkg < 0 || pkg > table {
 		t.Fatalf("nftables is not declared before the guard's table (%d, %d)", pkg, table)
 	}
-	if !reflect.DeepEqual(got[GuardRunningID()]["restart-on"], []any{GuardID(), GuardUnitID()}) {
-		t.Fatalf("the guard is not reloaded when its table changes: %v", got[GuardRunningID()])
+	// A changed table is reloaded — one `nft -f`, atomic — never restarted, which would delete the
+	// table and leave the ports unguarded until it is loaded again. Only a changed unit restarts.
+	if !reflect.DeepEqual(got[GuardRunningID()]["reload-on"], []any{GuardID()}) ||
+		!reflect.DeepEqual(got[GuardRunningID()]["restart-on"], []any{GuardUnitID()}) {
+		t.Fatalf("the guard is not reloaded on its table and restarted on its unit: %v",
+			got[GuardRunningID()])
 	}
 	// Nothing of the mesh's own is anybody's to hold.
 	for id, module := range composed.Owner {
@@ -317,6 +321,14 @@ func TestAGivenPortIsTheNodesAndReachesSomething(t *testing.T) {
 	if _, err := GivenPorts(store, node(map[string]any{"5432": float64(70000)})); err == nil {
 		t.Fatal("a machine port that is not a port was given")
 	}
+	if _, err := GivenPorts(store, node(map[string]any{"5432": float64(22)})); err == nil {
+		t.Fatal("ssh's port was given")
+	}
+	broker := anAdoptedAnchor().Modules[2]
+	if _, err := GivenPorts(broker, node(map[string]any{"5671": float64(5700),
+		"5672": float64(5700)})); err == nil {
+		t.Fatal("one machine port was given for two of the module's ports")
+	}
 	if stray := UnusedSettings(store, node(map[string]any{"5432": float64(5433)})); len(stray) != 0 {
 		t.Fatalf("a given port is called stray: %v", stray)
 	}

internal/catalogue/filtering.go

@@ -518,9 +518,22 @@ func GivenPorts(m Manifest, layers []Layer) (map[int]int, error) {
 				return nil, fmt.Errorf("%s gives port %d the machine port %v, which is not a port",
 					m.Module, port, value)
 			}
+			if at == SSHPort {
+				return nil, fmt.Errorf("%s gives port %d the machine port %d, which is ssh's — the "+
+					"one port a machine may never lose", m.Module, port, at)
+			}
 			out[port] = at
 		}
 	}
+	// One holder per machine port, within the module too.
+	holder := map[int]int{}
+	for port, at := range out {
+		if other, twice := holder[at]; twice {
+			return nil, fmt.Errorf("%s gives machine port %d to both its %d and its %d", m.Module,
+				at, min(port, other), max(port, other))
+		}
+		holder[at] = port
+	}
 	if len(out) == 0 {
 		return nil, nil
 	}

internal/catalogue/foundation_manifests_test.go

@@ -1,6 +1,7 @@
 package catalogue
 
 import (
+	"fmt"
 	"os"
 	"reflect"
 	"strings"
@@ -34,12 +35,13 @@ func TestTheStoreAndTheBrokerSayWhatTheMeshGuards(t *testing.T) {
 
 func TestTheFilterModuleNeverFlushesTheRuleset(t *testing.T) {
 	m := catalogueManifest(t, "nftables")
-	var unit map[string]any
-	var load map[string]any
+	var unit, stock, load map[string]any
 	for _, r := range m.Resources {
 		switch r["id"] {
 		case "unit":
 			unit = r
+		case "stock-unit-stop":
+			stock = r
 		case "load":
 			load = r
 		}
@@ -60,7 +62,16 @@ func TestTheFilterModuleNeverFlushesTheRuleset(t *testing.T) {
 		t.Fatalf("the unit does not load the computed rule set and delete only its own table:\n%s",
 			content)
 	}
-	if !reflect.DeepEqual(load["restart-on"], []any{"filtering", "unit"}) {
-		t.Fatalf("the filter is not reloaded when its rules or its unit change: %v", load["restart-on"])
+	// A node converged before the filter had its own unit still has the stock nftables.service
+	// enabled, whose stop flushes the whole ruleset: a drop-in makes it delete only the mesh's
+	// table, and the load is restarted on it so the host reloads units and the drop-in is read.
+	if stock == nil || stock["path"] != "/etc/systemd/system/nftables.service.d/mesh.conf" ||
+		!strings.HasSuffix(fmt.Sprint(stock["content"]),
+			"[Service]\nExecStop=\nExecStop=nft delete table inet mesh\n") {
+		t.Fatalf("the stock unit's stop is not replaced with deleting the mesh's table: %v", stock)
+	}
+	if !reflect.DeepEqual(load["restart-on"], []any{"filtering", "unit", "stock-unit-stop"}) {
+		t.Fatalf("the filter is not reloaded when its rules, its unit or the stock unit's drop-in "+
+			"change: %v", load["restart-on"])
 	}
 }

internal/inventory/catalogue.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
+	"strconv"
 	"strings"
 
 	"github.com/jackc/pgx/v5"
@@ -593,11 +595,157 @@ func (i *Inventory) SetSettings(ctx context.Context, nodeName, module string, va
 	if err != nil {
 		return err
 	}
-	_, err = i.store.Pool().Exec(ctx,
+	given, err := givenIn(module, raw)
+	if err != nil {
+		return err
+	}
+	tx, err := i.store.Pool().Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+	if len(given) > 0 {
+		if err := refuseGivenCollisions(ctx, tx, node.ID, nodeName, module, given); err != nil {
+			return err
+		}
+	}
+	_, err = tx.Exec(ctx,
 		`insert into settings (node, module, values) values ($1, $2, $3)
 		 on conflict (node, module) where node is not null
 		 do update set values = excluded.values, set_at = now()`, node.ID, module, raw)
-	return wrapModule(err, module)
+	if err != nil {
+		return wrapModule(err, module)
+	}
+	// A given port replaces what the mesh assigned for that port: the assignment is given back,
+	// so the number is free for the next module rather than held for ever in the name of a port
+	// that now lives elsewhere.
+	for wanted := range given {
+		if _, err := tx.Exec(ctx,
+			`delete from port_assignment where node = $1 and module = $2 and wanted = $3`,
+			node.ID, module, wanted); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
+}
+
+// givenIn is the machine ports a node-level settings layer gives a module, software port →
+// machine port (novox/hq ADR 0100). Nothing when the layer gives none; what is not a port is left
+// for composition to refuse in its own words.
+func givenIn(module string, raw []byte) (map[int]int, error) {
+	var layer map[string]any
+	if err := json.Unmarshal(raw, &layer); err != nil {
+		return nil, err
+	}
+	entries, ok := layer[catalogue.PortsSetting].(map[string]any)
+	if !ok {
+		return nil, nil
+	}
+	out := map[int]int{}
+	by := map[int]int{}
+	for text, value := range entries {
+		wanted, err := strconv.Atoi(text)
+		if err != nil {
+			continue
+		}
+		at, ok := value.(float64)
+		if !ok {
+			continue
+		}
+		machine := int(at)
+		if machine == catalogue.SSHPort {
+			return nil, fmt.Errorf("%s cannot be given port %d for its %d: that is ssh's, the one "+
+				"port a machine may never lose", module, machine, wanted)
+		}
+		if other, twice := by[machine]; twice {
+			return nil, fmt.Errorf("%s gives machine port %d to both its %d and its %d; a machine "+
+				"port has one holder", module, machine, min(other, wanted), max(other, wanted))
+		}
+		by[machine] = wanted
+		out[wanted] = machine
+	}
+	return out, nil
+}
+
+// refuseGivenCollisions refuses a given machine port another module on the node already has —
+// assigned by the mesh or given by its own setting — or that this module has for another of its
+// ports. A port it was assigned for the same software port is not a collision: the given one
+// replaces it.
+func refuseGivenCollisions(ctx context.Context, tx pgx.Tx, nodeID any, node, module string,
+	given map[int]int) error {
+	type holder struct {
+		module string
+		wanted int
+	}
+	held := map[int]holder{}
+	rows, err := tx.Query(ctx,
+		`select machine, module, wanted from port_assignment where node = $1`, nodeID)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var h holder
+		var machine int
+		if err := rows.Scan(&machine, &h.module, &h.wanted); err != nil {
+			rows.Close()
+			return err
+		}
+		held[machine] = h
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	rows, err = tx.Query(ctx,
+		`select module, values->'ports' from settings
+		  where node = $1 and module <> $2 and jsonb_typeof(values->'ports') = 'object'`,
+		nodeID, module)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var other string
+		var raw []byte
+		if err := rows.Scan(&other, &raw); err != nil {
+			rows.Close()
+			return err
+		}
+		var theirs map[string]any
+		if err := json.Unmarshal(raw, &theirs); err != nil {
+			rows.Close()
+			return err
+		}
+		for text, v := range theirs {
+			wanted, _ := strconv.Atoi(text)
+			if at, ok := v.(float64); ok {
+				held[int(at)] = holder{module: other, wanted: wanted}
+			}
+		}
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
+
+	wanted := make([]int, 0, len(given))
+	for w := range given {
+		wanted = append(wanted, w)
+	}
+	sort.Ints(wanted)
+	for _, w := range wanted {
+		machine := given[w]
+		h, taken := held[machine]
+		if !taken || (h.module == module && h.wanted == w) {
+			continue
+		}
+		if _, moving := given[h.wanted]; h.module == module && moving {
+			// Its own port for another of its software ports, which this same layer moves away.
+			continue
+		}
+		return fmt.Errorf("%w: %s cannot be given %d on %s for its %d — %s already has it for "+
+			"its %d", ErrPortTaken, module, machine, node, w, h.module, h.wanted)
+	}
+	return nil
 }
 
 func wrapModule(err error, module string) error {

internal/inventory/hold.go

@@ -0,0 +1,89 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+)
+
+// ErrNodeBusy is a node another act kept holding for longer than a caller waits.
+var ErrNodeBusy = errors.New("another act is composing or sending that node's declaration")
+
+// HoldWaitFor is how long HoldNodes waits for a node another act holds, and HoldPoll how often it
+// looks again. Variables so a test need not wait minutes.
+var (
+	HoldWaitFor = 2 * time.Minute
+	HoldPoll    = 250 * time.Millisecond
+)
+
+// HoldNodes serialises composing and sending a declaration per node (novox/hq ADR 0100): while
+// one caller holds a node, another asking for it waits. Without it a push that composed a node as
+// adopted could send that declaration after `converge --yes` sent the converged one, and the node
+// would return to adopted with nobody having asked.
+//
+// Session-level advisory locks on one connection, all or none: a set not wholly free is given back
+// at once, so two callers holding overlapping sets never each wait on the other. **A waiter pins
+// no connection.** It looks again every HoldPoll with a connection borrowed for the look, and gives
+// up after HoldWaitFor with ErrNodeBusy naming the node — so a stuck holder costs the pool one
+// connection, never one per caller queued behind it. Release gives every one back, and may be
+// called more than once.
+func (i *Inventory) HoldNodes(ctx context.Context, names []string) (func(), error) {
+	sorted := slices.Clone(names)
+	slices.Sort(sorted)
+	sorted = slices.Compact(sorted)
+	deadline := time.Now().Add(HoldWaitFor)
+	for {
+		release, busy, err := i.tryHold(ctx, sorted)
+		if err != nil || busy == "" {
+			return release, err
+		}
+		if time.Now().After(deadline) {
+			return nil, fmt.Errorf("%w: %s has been held for over %s — try again once it is done",
+				ErrNodeBusy, busy, HoldWaitFor)
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(HoldPoll):
+		}
+	}
+}
+
+// tryHold takes every named node's lock or none, and says which node was busy when it took none.
+func (i *Inventory) tryHold(ctx context.Context, sorted []string) (func(), string, error) {
+	conn, err := i.store.Pool().Acquire(ctx)
+	if err != nil {
+		return nil, "", err
+	}
+	var once sync.Once
+	release := func() {
+		once.Do(func() {
+			// Unlocking all of this session's advisory locks, then handing the connection back: a
+			// connection returned still holding one would hold it for whoever borrows it next.
+			_, err := conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock_all()`)
+			if err != nil {
+				// The session's locks die with the session: close it rather than return it.
+				_ = conn.Conn().Close(context.WithoutCancel(ctx))
+			}
+			conn.Release()
+		})
+	}
+	for _, name := range sorted {
+		var took bool
+		if err := conn.QueryRow(ctx,
+			`select pg_try_advisory_lock(hashtext('mesh-node-declaration:' || $1)::bigint)`,
+			name).Scan(&took); err != nil {
+			release()
+			return nil, "", err
+		}
+		if !took {
+			release()
+			return nil, strings.TrimSpace(name), nil
+		}
+	}
+	return release, "", nil
+}

internal/inventory/hold_test.go

@@ -0,0 +1,88 @@
+package inventory
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Composing and sending one node's declaration is serialised: a second holder waits for the first.
+func TestHoldingANodeMakesTheNextHolderWait(t *testing.T) {
+	inv := fresh(t)
+	ctx := t.Context()
+	release, err := inv.HoldNodes(ctx, []string{"anchor", "laptop"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := make(chan func(), 1)
+	go func() {
+		second, err := inv.HoldNodes(ctx, []string{"laptop"})
+		if err != nil {
+			t.Error(err)
+			got <- func() {}
+			return
+		}
+		got <- second
+	}()
+	select {
+	case <-got:
+		t.Fatal("a node held by one caller was held by another at the same time")
+	case <-time.After(300 * time.Millisecond):
+	}
+	// Another node is not held up.
+	other, err := inv.HoldNodes(ctx, []string{"joiner"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	other()
+	release()
+	select {
+	case second := <-got:
+		second()
+	case <-time.After(5 * time.Second):
+		t.Fatal("releasing the node did not let the next holder in")
+	}
+}
+
+// A waiter pins no pool connection while it waits, and gives up after a bounded wait saying which
+// node is busy.
+func TestAWaiterPinsNoConnectionAndGivesUp(t *testing.T) {
+	inv := fresh(t)
+	ctx := t.Context()
+	release, err := inv.HoldNodes(ctx, []string{"anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer release()
+	savedWait, savedPoll := HoldWaitFor, HoldPoll
+	HoldWaitFor, HoldPoll = 1500*time.Millisecond, 50*time.Millisecond
+	defer func() { HoldWaitFor, HoldPoll = savedWait, savedPoll }()
+
+	pool := inv.store.Pool()
+	base := pool.Stat().AcquiredConns()
+	const waiters = 3
+	done := make(chan error, waiters)
+	for range waiters {
+		go func() {
+			_, err := inv.HoldNodes(ctx, []string{"anchor"})
+			done <- err
+		}()
+	}
+	// While they wait, the pool lends nothing to them for longer than a look.
+	pinned := 0
+	for range 10 {
+		time.Sleep(60 * time.Millisecond)
+		if n := int(pool.Stat().AcquiredConns() - base); n > pinned {
+			pinned = n
+		}
+	}
+	if pinned >= waiters {
+		t.Fatalf("%d connections were held by %d waiters", pinned, waiters)
+	}
+	for range waiters {
+		if err := <-done; !errors.Is(err, ErrNodeBusy) || !strings.Contains(err.Error(), "anchor") {
+			t.Fatalf("a waiter did not give up naming the busy node: %v", err)
+		}
+	}
+}

internal/inventory/ports_test.go

@@ -257,3 +257,64 @@ func TestAGivenPortIsNeverAssigned(t *testing.T) {
 		t.Fatalf("a fixed port given to another module was handed over: %v", err)
 	}
 }
+
+// novox/hq ADR 0100: a given machine port has one holder. It is refused when another module has it,
+// assigned or given, when the same layer gives it twice, and when it is ssh's; and when it replaces
+// what the mesh assigned for that port, the assignment is given back.
+func TestAGivenPortHasOneHolderAndReplacesTheAssignment(t *testing.T) {
+	inv, node := aNodeWithModules(t, "postgres", "web", "cache")
+	ctx := t.Context()
+	give := func(module string, ports map[string]any) error {
+		return inv.SetSettings(ctx, node, module, map[string]any{catalogue.PortsSetting: ports})
+	}
+
+	web, err := inv.PortFor(ctx, node, "web", 8080, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := give("postgres", map[string]any{"5432": web.Machine}); !errors.Is(err, ErrPortTaken) {
+		t.Fatalf("a port the mesh assigned to web was given to postgres: %v", err)
+	}
+	if err := give("postgres", map[string]any{"5432": 22}); err == nil {
+		t.Fatal("ssh's port was given")
+	}
+	if err := give("postgres", map[string]any{"5432": 5433, "5433": 5433}); err == nil {
+		t.Fatal("one machine port was given for two ports")
+	}
+	if err := give("cache", map[string]any{"6379": 6380}); err != nil {
+		t.Fatal(err)
+	}
+	if err := give("postgres", map[string]any{"5432": 6380}); !errors.Is(err, ErrPortTaken) {
+		t.Fatalf("a port given to cache was given to postgres: %v", err)
+	}
+
+	// Postgres was assigned a port for 5432; given one, the assignment is released and the number
+	// is free again.
+	assigned, err := inv.PortFor(ctx, node, "postgres", 5432, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := give("postgres", map[string]any{"5432": 5433}); err != nil {
+		t.Fatal(err)
+	}
+	// Given again, the same: its own given port is not a collision with itself.
+	if err := give("postgres", map[string]any{"5432": 5433}); err != nil {
+		t.Fatal(err)
+	}
+	held, err := inv.PortsFor(ctx, node)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, a := range held {
+		if a.Module == "postgres" {
+			t.Fatalf("the assignment a given port replaced is still held: %+v", a)
+		}
+	}
+	other, err := inv.PortFor(ctx, node, "cache", 11211, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if other.Machine != assigned.Machine {
+		t.Fatalf("the released port %d was not free again (got %d)", assigned.Machine, other.Machine)
+	}
+}