diff --git a/go.mod b/go.mod index 26d0523..37f75e4 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index e14ad38..1bb95c5 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/link/have.go b/internal/link/have.go index 99ef52b..0d59e19 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -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,23 @@ 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 } + for module, build := range s.written { + 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 +172,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 +216,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 +242,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 { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 90f465b..516dee3 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -339,3 +339,91 @@ 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) + } +}