2 Commits
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
6 changed files with 223 additions and 6 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"