Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
150fadfe8b | ||
|
|
6a2fb63bfc |
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user