Author SHA1 Message Date
mesh-tools 150fadfe8b Write a removal after a restart from the declaration the apply replaced (issue 1369)
mesh/repo-check head 150fadfe8bed merged into trunk 2474bf28330f — pass: its merge-check.sh passed
mesh/delivery ready: it delivers once merged
mesh/merge-gate head 150fadfe8bed merged into trunk 2474bf28330f — pass: group fix/r3e-before-the-gate composed: every machine composes with the change as …
mesh/delivery-group group fix/r3e-before-the-gate ready: every member ready, and composed together they pass
The set of modules this process wrote lives in memory, so a node-engine restarted before the apply that drops a module never wrote that module's removal. The report now names the modules the kept declaration held before, and they are removal candidates.
2026-10-20 01:26:33 +00:00
mesh-tools 6a2fb63bfc Write a module's have key as removed when the applied declaration no longer holds it (issue 1369)
A module that left a node kept its have key at running for ever, so nothing could tell it stopped and the gate would never take the seat. The key is written once with the phase removed over the revision read, never deleted. Pins mesh-sdk to the commit that adds the phase.
2026-10-20 01:20:52 +00:00
9 changed files with 223 additions and 228 deletions
+11
View File
@@ -2314,7 +2314,17 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
}
// Kept whichever way it went, so a node that is disconnected next minute still knows what it
// was told. Saved after applying rather than before: what is kept is what this node acted on.
var before []string
if signed != nil {
// The modules the declaration kept until now held, so a have key written for one the new declaration no longer
// holds is written removed even by a node-engine that started since (issue 1369).
if kept, keptErr := store.ReadDeclared(store.DeclaredPath(opts.state)); keptErr == nil {
if old, parseErr := declaration.ParseTrusted(kept.Declaration); parseErr == nil {
for _, f := range moduleFacts(old, apply.Report{}, nil) {
before = append(before, f.Module)
}
}
}
if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil {
report.Failed = map[string]string{"keeping the declaration": err.Error()}
}
@@ -2326,6 +2336,7 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
// And how each module of the declaration came out, for the have keys written beside this report (novox/hq ADR 0429
// §4, issue 1132): not on the wire.
report.Modules = moduleFacts(declared, outcome, applyErr)
report.Before = before
// Which resources did not apply, for whether a reconcile's account is news (issue 233): not said on the wire.
report.Unapplied = unapplied(declared, outcome, applyErr)
return report
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020011849-ba684e54976b
github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0
golang.org/x/sys v0.48.0
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 h1:BrHyE3ifFMUQLIza6I/iNlEgdcysB9/Y+1FO6Q9xws8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020011849-ba684e54976b h1:U7++vBnHNWRYpwdbO+Tx03HUcEzDFLBciNDe0fNiXYc=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020011849-ba684e54976b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+95 -3
View File
@@ -37,6 +37,11 @@ type haveState struct {
modules map[string]ModuleFact
health map[string]delivery.Health
node *delivery.HaveNode
// written is the build of each module this node-engine wrote a have key for and that the declaration it applied
// still holds; removed is each such module the latest declaration no longer holds, with the build its key last
// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
written map[string]string
removed map[string]string
}
// moduleHealths is the health of each module a look names, from the states of its resources: failed when any is
@@ -99,10 +104,39 @@ func (q *Queue) noteApply(r Report) bool {
s := &q.haves
s.mu.Lock()
defer s.mu.Unlock()
s.modules = map[string]ModuleFact{}
next := map[string]ModuleFact{}
for _, m := range r.Modules {
s.modules[m.Module] = m
next[m.Module] = m
}
// Candidates: what this process wrote, what the declaration kept before this one held (a node-engine that started
// since wrote nothing yet), and what the last apply here held. writeRemoved creates no missing key.
gone := map[string]string{}
for module, build := range s.written {
gone[module] = build
}
for _, module := range r.Before {
if _, known := gone[module]; !known {
gone[module] = ""
}
}
for module, f := range s.modules {
if _, known := gone[module]; !known {
gone[module] = f.Build
}
}
for module, build := range gone {
if _, held := next[module]; !held {
if s.removed == nil {
s.removed = map[string]string{}
}
s.removed[module] = build
delete(s.written, module)
}
}
for module := range next {
delete(s.removed, module) // added back: its have is written with its new phase
}
s.modules = next
if r.Health != nil {
s.health = moduleHealths(r.Health)
}
@@ -154,7 +188,7 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
s := &q.haves
s.mu.Lock()
defer s.mu.Unlock()
if s.modules == nil && s.node == nil {
if s.modules == nil && s.node == nil && len(s.removed) == 0 {
return
}
node := q.Membership.Node
@@ -198,6 +232,13 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
}) {
return
}
if s.written == nil {
s.written = map[string]string{}
}
s.written[module] = f.Build
}
if !q.writeRemoved(write, bucket, node) {
return
}
if s.node != nil {
key, err := delivery.HaveNodeKey(node)
@@ -217,6 +258,57 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
}
}
// writeRemoved writes, once, each module the applied declaration no longer holds and this node-engine wrote a have key
// for, with the phase removed over the revision it reads (issue 1369). The key is never deleted, and a key that is not
// in the bucket is not created. False when the pass should stop. The caller holds the lock of the haves.
func (q *Queue) writeRemoved(ctx context.Context, b HaveBucket, node string) bool {
s := &q.haves
gone := make([]string, 0, len(s.removed))
for module := range s.removed {
gone = append(gone, module)
}
sort.Strings(gone)
for _, module := range gone {
key, err := delivery.HaveKey(node, module)
if err != nil {
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
delete(s.removed, module)
continue
}
h := delivery.Have{V: delivery.Version, Build: s.removed[module], Phase: delivery.PhaseRemoved,
Health: delivery.HealthUnknown, Detail: "the applied declaration no longer holds the module"}
raw, err := delivery.EncodeHave(h)
if err != nil {
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
delete(s.removed, module)
continue
}
if ctx.Err() != nil {
q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
return false
}
current, revision, err := b.Get(ctx, key)
switch {
case errors.Is(err, ErrHaveAbsent):
delete(s.removed, module) // nothing to say: the key is not there
continue
case err != nil:
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
return false
}
if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved {
delete(s.removed, module)
continue
}
if err := b.Set(ctx, key, raw, revision); err != nil {
q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
return false
}
delete(s.removed, module)
}
return true
}
// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the
// bucket failed, the error is said once, and the rest wait for the next pass.
func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool {
+109
View File
@@ -339,3 +339,112 @@ func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said)
}
}
func onlyA() Report {
return Report{Declared: "d2", Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1"}},
Health: healthOf(map[string]string{"a": StateHealthy})}
}
// 1369/AC1: after a declaration with a and b, one with a only writes n1.b with the phase removed, over the revision
// read, and leaves a as it was; the key is not deleted.
func TestAModuleLeavingTheDeclarationIsWrittenRemoved(t *testing.T) {
l := newHaveLink()
q, _ := aHaveQueue(l)
q.tell(context.Background(), appliedModules(), applied)
revA, revB := l.kv["n1.a"].rev, l.kv["n1.b"].rev
q.tell(context.Background(), onlyA(), applied)
b := l.have(t, "n1.b")
if b.Phase != delivery.PhaseRemoved || b.Build != "b-2" {
t.Errorf("b: %+v", b)
}
if l.kv["n1.b"].rev != revB+1 {
t.Errorf("b was written at revision %d, want one over %d", l.kv["n1.b"].rev, revB)
}
if l.kv["n1.a"].rev != revA || l.have(t, "n1.a").Phase != delivery.PhaseRunning {
t.Errorf("a changed: %+v at %d", l.have(t, "n1.a"), l.kv["n1.a"].rev)
}
before := l.setCount()
q.tell(context.Background(), onlyA(), applied)
if l.setCount() != before {
t.Errorf("a second apply without b wrote %d keys, removed is written once", l.setCount()-before)
}
}
// 1369/AC2: a module removed and later added back is written with its new phase.
func TestARemovedModuleAddedBackIsWrittenWithItsNewPhase(t *testing.T) {
l := newHaveLink()
q, _ := aHaveQueue(l)
q.tell(context.Background(), appliedModules(), applied)
q.tell(context.Background(), onlyA(), applied)
q.tell(context.Background(), appliedModules(), applied)
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseFailed || b.Build != "b-2" {
t.Errorf("b after it came back: %+v", b)
}
q.tell(context.Background(), onlyA(), applied)
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
t.Errorf("b removed again: %+v", b)
}
}
// 1369/AC3: a module this node-engine never wrote a have key for is not written, and a key that is gone from the
// bucket is not created again.
func TestAModuleNeverWrittenIsNotWrittenRemoved(t *testing.T) {
l := newHaveLink()
l.put(t, "n1.c", delivery.Have{V: delivery.Version, Build: "b-9", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
q, _ := aHaveQueue(l)
q.tell(context.Background(), appliedModules(), applied)
q.tell(context.Background(), onlyA(), applied)
if c := l.have(t, "n1.c"); c.Phase != delivery.PhaseRunning || l.kv["n1.c"].rev != 1 {
t.Errorf("c, which this node-engine never wrote, changed: %+v at %d", c, l.kv["n1.c"].rev)
}
// a key that vanished from the bucket is not created by the removal
l2 := newHaveLink()
q2, _ := aHaveQueue(l2)
q2.tell(context.Background(), appliedModules(), applied)
l2.hmu.Lock()
delete(l2.kv, "n1.b")
l2.hmu.Unlock()
q2.tell(context.Background(), onlyA(), applied)
if _, there := l2.kv["n1.b"]; there {
t.Error("the removal created a key that was not there")
}
}
// A removal the bucket refused is written by the next pass.
func TestARefusedRemovalIsWrittenByTheNextPass(t *testing.T) {
l := newHaveLink()
q, _ := aHaveQueue(l)
q.tell(context.Background(), appliedModules(), applied)
l.hmu.Lock()
l.failing = true
l.hmu.Unlock()
q.tell(context.Background(), onlyA(), applied)
l.hmu.Lock()
l.failing = false
l.hmu.Unlock()
q.tell(context.Background(), onlyA(), applied)
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
t.Errorf("b: %+v", b)
}
}
// 1369/F1: a node-engine that started since wrote nothing, and the report names the modules the kept declaration held
// before: one of them the new declaration lacks is written removed, one that is not in the bucket is not created.
func TestARestartedNodeEngineWritesTheRemovalFromTheReportsBefore(t *testing.T) {
l := newHaveLink()
l.put(t, "n1.a", delivery.Have{V: delivery.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.b", delivery.Have{V: delivery.Version, Build: "b-2", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
q, _ := aHaveQueue(l) // fresh: nothing written by this process
r := onlyA()
r.Before = []string{"a", "b", "c"}
q.tell(context.Background(), r, applied)
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
t.Errorf("b: %+v", b)
}
if _, there := l.kv["n1.c"]; there {
t.Error("c, which has no key, was created")
}
if l.kv["n1.a"].rev != 1 {
t.Errorf("a was written, revision %d", l.kv["n1.a"].rev)
}
}
+5
View File
@@ -188,6 +188,11 @@ type Report struct {
// have. Nil on a report that is not an apply's.
Modules []ModuleFact `json:"-"`
// Before is the names of the modules the declaration kept before this apply held, taken before it was replaced
// (issue 1369): a module of it that Modules lacks has left the node, whether or not this process wrote its have key.
// **Not on the wire.**
Before []string `json:"-"`
// Host is the version of the host that produced this report (novox/hq ADR 0141).
//
// Without it nothing can say a machine is behind, so "every machine current with its source"
-6
View File
@@ -24,11 +24,6 @@ const (
CapSeat = "seat"
CapPrivileged = "privileged"
// CapUserNamespaces is whether an unprivileged user can make a user namespace: what a build
// module's sandbox (bubblewrap) needs (novox/hq ADR 0448 section 5). Proven by making one, as
// an unprivileged user, never by a version string or a sysctl alone.
CapUserNamespaces = "user-namespaces"
// The network manager this machine runs, one capability per dialect (novox/hq ADR 0161): the
// uplink seat's holder declares its own, so the holder for a manager the machine does not run
// is refused the way any missing capability is, naming it. Active, not installed — a machine
@@ -194,7 +189,6 @@ func Default(runner Runner) []Detector {
seat{},
battery{},
powerMeter{},
userNamespaces(runner),
commandCapability{
name: CapContainerRuntime, command: "docker", args: []string{"info", "--format", "{{.ServerVersion}}"},
why: "asks the daemon for its version — a running daemon, not an installed client",
-61
View File
@@ -1,61 +0,0 @@
package profile
import (
"fmt"
"os"
"strings"
)
// effectiveUID and readProcFile are replaceable so the unit tests stay off the machine.
var (
effectiveUID = os.Geteuid
readProcFile = func(path string) (string, error) {
b, err := os.ReadFile(path)
return string(b), err
}
)
// unprivilegedID is the uid and gid the probe drops to (nobody on every mesh node).
const unprivilegedID = "65534"
// userNamespaces builds the detector for CapUserNamespaces: it makes a user namespace as an
// unprivileged user. The node-engine runs as root, and root can always make one, so as root the
// probe drops first with setpriv (util-linux). Go's SysProcAttr Cloneflags with a Credential
// would clone before the uid drops and prove nothing, so it is not used.
func userNamespaces(runner Runner) Detector {
command, args := "unshare", []string{"--user", "true"}
if effectiveUID() == 0 {
command = "setpriv"
args = []string{"--reuid=" + unprivilegedID, "--regid=" + unprivilegedID, "--clear-groups", "unshare", "--user", "true"}
}
read := readProcFile // bound now, so a probe cut by the timeout never reads a later replacement
return commandCapability{
name: CapUserNamespaces, command: command, args: args,
why: "makes a user namespace as an unprivileged user — what bubblewrap needs, not a kernel setting read",
interpret: func(_ string, err error) (bool, string) {
if err == nil {
return true, "an unprivileged user can make a user namespace"
}
return false, err.Error() + userNamespaceSettings(read)
},
runner: runner,
}
}
// userNamespaceSettings names the two kernel settings that switch user namespaces off, when
// they can be read. It is added to the reason of an absent verdict only.
func userNamespaceSettings(read func(string) (string, error)) string {
var parts []string
for _, s := range []struct{ path, name string }{
{"/proc/sys/kernel/unprivileged_userns_clone", "unprivileged_userns_clone"},
{"/proc/sys/user/max_user_namespaces", "max_user_namespaces"},
} {
if v, err := read(s.path); err == nil {
parts = append(parts, fmt.Sprintf("%s=%s", s.name, strings.TrimSpace(v)))
}
}
if len(parts) == 0 {
return ""
}
return " (" + strings.Join(parts, ", ") + ")"
}
-155
View File
@@ -1,155 +0,0 @@
package profile
import (
"context"
"errors"
"strings"
"sync"
"testing"
"time"
)
// recorded is a runner that remembers every command it was given and answers the user-namespace
// probe with the given output and error. Other detectors get "not here".
type recorded struct {
mu sync.Mutex
calls [][]string
out string
err error
}
func (r *recorded) run(_ context.Context, name string, args ...string) (string, error) {
if name != "setpriv" && name != "unshare" {
return "", errors.New("not here")
}
r.mu.Lock()
defer r.mu.Unlock()
r.calls = append(r.calls, append([]string{name}, args...))
return r.out, r.err
}
// asUID makes the detector believe it runs as uid, and reads no files from the machine.
func asUID(t *testing.T, uid int, procs map[string]string) {
t.Helper()
oldUID, oldRead := effectiveUID, readProcFile
effectiveUID = func() int { return uid }
readProcFile = func(path string) (string, error) {
if v, ok := procs[path]; ok {
return v, nil
}
return "", errors.New("no such file")
}
t.Cleanup(func() { effectiveUID, readProcFile = oldUID, oldRead })
}
func verdictOf(t *testing.T, p Profile, name string) Verdict {
t.Helper()
for _, v := range p.Capabilities {
if v.Name == name {
return v
}
}
t.Fatalf("no verdict named %s in the profile", name)
return Verdict{}
}
// unshare's own words when user namespaces are off, from util-linux's unshare.c:
// err(EXIT_FAILURE, _("unshare failed")), which prints "unshare: unshare failed: " and strerror(EPERM).
// ExecRunner wraps stderr as "<name> exited <code>: <stderr>".
const unshareRefused = "unshare exited 1: unshare: unshare failed: Operation not permitted"
func TestUserNamespacesAsRootDropsToAnUnprivilegedUIDFirst(t *testing.T) {
// 1308/AC1. Root can always make a user namespace, so a probe run as root proves nothing: the
// command must drop to an unprivileged uid before unshare runs.
asUID(t, 0, nil)
r := &recorded{}
got := Detect(context.Background(), Default(r.run), time.Second)
v := verdictOf(t, got, CapUserNamespaces)
if !v.Present {
t.Fatalf("a successful probe was reported absent: %+v", v)
}
want := "setpriv --reuid=65534 --regid=65534 --clear-groups unshare --user true"
if len(r.calls) != 1 || strings.Join(r.calls[0], " ") != want {
t.Fatalf("the probe ran %v, want one call: %s", r.calls, want)
}
}
func TestUserNamespacesAbsentCarriesTheCommandsWords(t *testing.T) {
// 1308/AC2. The failing probe gives absent, and the detail is what unshare said.
asUID(t, 0, nil)
r := &recorded{err: errors.New(unshareRefused)}
got := Detect(context.Background(), Default(r.run), time.Second)
v := verdictOf(t, got, CapUserNamespaces)
if v.Present {
t.Fatal("a probe that failed was reported present")
}
if !strings.Contains(v.Detail, "unshare failed: Operation not permitted") {
t.Fatalf("the command's words were lost; detail was %q", v.Detail)
}
if strings.TrimSpace(v.How) == "" {
t.Fatal("an absent verdict with no stated method")
}
}
func TestUserNamespacesAbsentSaysWhyFromTheKernelSettings(t *testing.T) {
// 1308/AC2, the reasons: the two kernel settings that switch user namespaces off are read
// through a replaceable function and named in the detail.
asUID(t, 0, map[string]string{
"/proc/sys/user/max_user_namespaces": "0\n",
"/proc/sys/kernel/unprivileged_userns_clone": "0\n",
})
r := &recorded{err: errors.New(unshareRefused)}
v := verdictOf(t, Detect(context.Background(), Default(r.run), time.Second), CapUserNamespaces)
if v.Present {
t.Fatal("reported present")
}
// The detail is one line (firstLine), so the settings must be on that line.
if !strings.Contains(v.Detail, "max_user_namespaces=0") || !strings.Contains(v.Detail, "unprivileged_userns_clone=0") {
t.Fatalf("the kernel settings are not in the detail: %q", v.Detail)
}
}
func TestUserNamespacesNotAsRootUnsharesDirectly(t *testing.T) {
// 1308/AC3. An unprivileged node-engine is already the unprivileged user: no setpriv.
asUID(t, 1000, nil)
r := &recorded{}
v := verdictOf(t, Detect(context.Background(), Default(r.run), time.Second), CapUserNamespaces)
if !v.Present {
t.Fatalf("a successful probe was reported absent: %+v", v)
}
if len(r.calls) != 1 || strings.Join(r.calls[0], " ") != "unshare --user true" {
t.Fatalf("the probe ran %v, want: unshare --user true", r.calls)
}
if strings.TrimSpace(v.Detail) == "" || strings.TrimSpace(v.How) == "" {
t.Fatalf("present verdict lacks detail or how: %+v", v)
}
}
func TestUserNamespacesProbeThatHangsIsCutAndTheRestStillReport(t *testing.T) {
// 1308/AC4. A wedged unshare is cut by Detect's timeout; the other detectors still answer.
asUID(t, 0, nil)
blocking := func(ctx context.Context, name string, args ...string) (string, error) {
if name == "setpriv" {
<-ctx.Done()
return "", ctx.Err()
}
return "ok", nil
}
done := make(chan Profile, 1)
go func() { done <- Detect(context.Background(), Default(blocking), 50*time.Millisecond) }()
select {
case got := <-done:
if verdictOf(t, got, CapUserNamespaces).Present {
t.Error("a probe that never answered was reported present")
}
if !got.Has(CapPackageManager) {
t.Error("the other capabilities were lost behind the wedged probe")
}
case <-time.After(5 * time.Second):
t.Fatal("detection did not return")
}
}