diff --git a/cmd/mesh-host/healthowed_test.go b/cmd/mesh-host/healthowed_test.go new file mode 100644 index 0000000..ba4907d --- /dev/null +++ b/cmd/mesh-host/healthowed_test.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "errors" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/liveness" +) + +// owedBus is a link that takes reports and statements and keeps a have bucket in memory. +type owedBus struct { + mu sync.Mutex + kv map[string][]byte + rv map[string]uint64 +} + +func (b *owedBus) Report(context.Context, string, []byte) error { return nil } +func (b *owedBus) Health(context.Context, string, []byte) error { return nil } +func (b *owedBus) HaveBind(context.Context) (link.HaveBucket, error) { return b, nil } + +func (b *owedBus) Get(_ context.Context, key string) ([]byte, uint64, error) { + b.mu.Lock() + defer b.mu.Unlock() + v, ok := b.kv[key] + if !ok { + return nil, 0, link.ErrHaveAbsent + } + return v, b.rv[key], nil +} + +func (b *owedBus) Set(_ context.Context, key string, value []byte, revision uint64) error { + b.mu.Lock() + defer b.mu.Unlock() + if b.rv[key] != revision { + return errors.New("wrong last sequence") + } + b.kv[key], b.rv[key] = value, revision+1 + return nil +} + +func (b *owedBus) have(key string) (delivery.Have, bool) { + b.mu.Lock() + defer b.mu.Unlock() + h, err := delivery.DecodeHave(b.kv[key]) + return h, err == nil +} + +// A module with nothing long-running (xorg) is ok within one look of its apply, not at the five-minute statement: the +// facts look inside the apply runs while it is applying and judges nothing, and the running write then owes a look, +// which the health loop makes at once (review of fix 15, B1). +func TestAModuleWithNothingLongRunningIsOkWithinALookOfItsApply(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + bus := &owedBus{kv: map[string][]byte{}, rv: map[string]uint64{}} + queue := &link.Session{Membership: link.Membership{Node: "n1"}, Timeout: time.Second} + queue.Attach(ctx, bus) + j, err := liveness.Open(filepath.Join(t.TempDir(), "liveness.json"), nil) + if err != nil { + t.Fatal(err) + } + fact := link.ModuleFact{Module: "xorg", FP: "src2:" + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("b", 64)} + applying := fact + applying.Phase = string(delivery.PhaseApplying) + queue.NoteModule(ctx, applying) + go judgeWhatRuns(ctx, j, queue, func(string) {}, nil) + // The facts look inside the apply, while the module is still applying. + queue.SayFacts(ctx, link.Report{Health: &link.Health{Contract: link.LivenessContract, At: time.Now().UTC()}}) + if h, _ := bus.have("n1.xorg"); h.Health != delivery.HealthUnknown { + t.Fatalf("judged while applying: %+v", h) + } + queue.NoteModule(ctx, fact) + deadline := time.Now().Add(2 * time.Second) // well within one look (liveness.LookEvery, 15 s) + for { + h, _ := bus.have("n1.xorg") + if h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK { + return + } + if time.Now().After(deadline) { + t.Fatalf("not ok within a look of its apply: %+v", h) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 7b13709..06ebfd9 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1340,11 +1340,19 @@ func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Session, var lastSaid time.Time owed := true spaced := 1.0 + // A module whose change ended unjudged is looked at and said at once (review of fix 15): a module with nothing + // long-running is then ok within a look of its apply, not at the five-minute statement. + var nudged <-chan struct{} + if queue != nil { + nudged = queue.HealthOwed() + } for { select { case <-ctx.Done(): return case <-ticker.C: + case <-nudged: + owed = true } st, changed := j.Look(ctx) owed = owed || changed diff --git a/internal/agent/agent.go b/internal/agent/agent.go index cefbd72..7c14c4a 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -671,6 +671,10 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com a.mu.Lock() a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning} a.mu.Unlock() + // Said again, so the session holds the module and its key carries the health this engine judges, + // not the one an older engine left there (xorg's unknown on g14 and shanks, 2026-10-25). + a.env.Haves.Write(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed, + Phase: string(delivery.PhaseRunning), Detail: h.Detail}) return true } return false diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 7319982..8709ee6 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -313,6 +313,40 @@ func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) { } } +// A module found running on its own have after a restart is said again, once, through the haves: the session then holds +// it and writes the health this engine judges. Not said, its key kept the health an older engine wrote: xorg's have on +// g14 and shanks said unknown from before the engine said a module with nothing to judge ok, and the assigner held every +// module requiring x11-display back for it (2026-10-25). +func TestStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + ck, craw, dig := composition(t, "xorg", fp, false, file("xorg", "x", "1")) + wk, wraw := wantFor(t, "xorg", fp, dig, "") + b.haves.stored = map[string]Have{"xorg": {Module: "xorg", FP: fp, Composed: dig, Phase: "running", Detail: "kept"}} + b.comps.before(ck, craw) + b.wants.before(wk, wraw) + stop := b.run() + defer stop() + eventually(t, "the have found running was not said again", func() bool { + b.haves.mu.Lock() + defer b.haves.mu.Unlock() + return len(b.haves.log) > 0 + }) + time.Sleep(50 * time.Millisecond) + b.haves.mu.Lock() + log := append([]Have(nil), b.haves.log...) + b.haves.mu.Unlock() + if len(log) != 1 { + t.Fatalf("the have was said %d times, want once: %+v", len(log), log) + } + if h := log[0]; h.Module != "xorg" || h.FP != fp || h.Composed != dig || h.Phase != string(delivery.PhaseRunning) || h.Detail != "kept" { + t.Fatalf("the have said again: %+v", h) + } + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a module that runs what the want names was applied: %v", got) + } +} + // A module whose kept resources differ from the composition is applied, and a failed apply is retried only by a change // of either key. func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { diff --git a/internal/link/have.go b/internal/link/have.go index 35ec1c0..d47fdb9 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -47,6 +47,74 @@ type haveState struct { // 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 + // unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look + // has judged since it stopped moving: its health is not yet judged, never the last look's of what ran before. + unjudged map[string]bool +} + +// inMotion says a module's phase is one of a change still under way: what a look finds of it then judges nothing of the +// composition being applied (a container stopped to be replaced reads unhealthy), so its health is not yet judged. +func inMotion(phase string) bool { + switch delivery.Phase(phase) { + case delivery.PhaseBackingUp, delivery.PhaseApplying, delivery.PhaseChecking: + return true + } + return false +} + +// healthOf is the health a module's have carries: unknown (not yet judged) while its change is under way or until a look +// after it, the last look's otherwise. The caller holds the lock. +func (s *haveState) healthOf(module string, f ModuleFact) delivery.Health { + if inMotion(f.Phase) { + return delivery.HealthUnknown + } + if got, known := s.health[module]; known { + return got + } + if s.unjudged[module] { + return delivery.HealthUnknown + } + return nothingJudged(f, s.looked) +} + +// moved marks a module's health not yet judged when its fact f is a change under way, or names another composition than +// the fact before it: the last look judged what ran before. The caller holds the lock. +func (s *haveState) moved(f ModuleFact) { + if prev, known := s.modules[f.Module]; inMotion(f.Phase) || known && (prev.FP != f.FP || prev.Composed != f.Composed) { + delete(s.health, f.Module) + if s.unjudged == nil { + s.unjudged = map[string]bool{} + } + s.unjudged[f.Module] = true + } +} + +// owed says some module left motion and no look judged it since. The caller holds the lock. +func (s *haveState) owed() bool { + for module, f := range s.modules { + if s.unjudged[module] && !inMotion(f.Phase) { + return true + } + } + return false +} + +// takeLook keeps a look's health: of a module still in motion nothing, since the look judged what is being replaced; +// of every other module what the look found, which is then judged. The caller holds the lock. +func (s *haveState) takeLook(h *Health) { + next := moduleHealths(h) + for module, f := range s.modules { + if inMotion(f.Phase) { + delete(next, module) + if s.unjudged == nil { + s.unjudged = map[string]bool{} + } + s.unjudged[module] = true + continue + } + delete(s.unjudged, module) + } + s.health, s.looked = next, true } // moduleHealths is the health of each module a look names, from the states of its resources: failed when any is @@ -113,7 +181,7 @@ func (q *Session) noteApply(r Report) bool { s.mu.Lock() defer s.mu.Unlock() if r.Health != nil { - s.health, s.looked = moduleHealths(r.Health), true + s.takeLook(r.Health) } n := nodeFacts(r) s.node = &n @@ -153,9 +221,20 @@ func (q *Session) noteApply(r Report) bool { for module := range next { delete(s.removed, module) // added back: its have is written with its new phase } + for module := range s.modules { + if _, held := next[module]; !held { + delete(s.unjudged, module) + } + } + for _, f := range next { + s.moved(f) + } s.modules = next if r.Health != nil { - s.health, s.looked = moduleHealths(r.Health), true + s.takeLook(r.Health) + } + if s.owed() { + q.oweHealth() } n := nodeFacts(r) s.node = &n @@ -167,22 +246,17 @@ func (q *Session) noteHealth(h Health) bool { s := &q.haves s.mu.Lock() defer s.mu.Unlock() - next := moduleHealths(&h) + before := map[string]delivery.Health{} + for module, f := range s.modules { + before[module] = s.healthOf(module, f) + } + s.takeLook(&h) changed := false for module, f := range s.modules { - want, ok := next[module] - if !ok { - want = nothingJudged(f, true) - } - have, known := s.health[module] - if !known { - have = nothingJudged(f, s.looked) - } - if have != want { + if s.healthOf(module, f) != before[module] { changed = true } } - s.health, s.looked = next, true return changed } @@ -233,11 +307,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { case f.Failed: h.Phase = delivery.PhaseFailed } - if got, known := s.health[module]; known { - h.Health = got - } else { - h.Health = nothingJudged(f, s.looked) - } + h.Health = s.healthOf(module, f) 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)) @@ -378,9 +448,18 @@ func (q *Session) NoteModule(ctx context.Context, f ModuleFact) { if s.modules == nil { s.modules = map[string]ModuleFact{} } + // A change under way, or a fact naming another composition than the one looked at, leaves the module's health not + // yet judged until a look after it: the last look judged what ran before (2026-10-25, the canary of the controller on + // novox held on a have saying applying and failed two seconds before its launch said healthy). + s.moved(f) s.modules[f.Module] = f delete(s.removed, f.Module) + // Out of motion and not judged: a look is owed now, not at the loop's next statement (review of fix 15, B1). + owe := !inMotion(f.Phase) && s.unjudged[f.Module] s.mu.Unlock() + if owe { + q.oweHealth() + } q.mu.Lock() bus := q.bus q.mu.Unlock() @@ -402,6 +481,7 @@ func (q *Session) NoteRemoved(ctx context.Context, module, build string) { s.removed[module] = build delete(s.modules, module) delete(s.written, module) + delete(s.unjudged, module) s.mu.Unlock() q.mu.Lock() bus := q.bus diff --git a/internal/link/have_test.go b/internal/link/have_test.go index a16b840..3a87c64 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -521,3 +521,101 @@ func TestAModuleWithNothingToJudgeIsOkOnceLookedAt(t *testing.T) { t.Fatalf("a failed module was said ok") } } + +// A have an older engine left unknown is written over with this engine's health once the module is said again, as the +// node agent says a module it finds running on its own have: a packages-only module (xorg, nothing long-running) is ok +// after a look, so the assigner no longer holds back what requires its provision (2026-10-25, g14 and shanks). +func TestAModuleSaidAgainOverAnOlderEnginesUnknownIsWrittenWithThisEnginesHealth(t *testing.T) { + l := newHaveLink() + fp := "src2:" + strings.Repeat("a", 64) + dig := "sha256:" + strings.Repeat("b", 64) + l.put(t, "n1.xorg", delivery.Have{V: delivery.Version, FP: fp, Composed: dig, Phase: delivery.PhaseRunning, + Health: delivery.HealthUnknown}) + q, _ := aHaveQueue(l) + if !q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) { + t.Fatal("not said") + } + if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthUnknown { + t.Fatalf("a module the session does not hold was written: %+v", h) + } + q.NoteModule(context.Background(), ModuleFact{Module: "xorg", FP: fp, Composed: dig}) + if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthOK || h.FP != fp || h.Composed != dig || h.Phase != delivery.PhaseRunning { + t.Fatalf("said again after a look: %+v", h) + } +} + +// A module's health is not yet judged (unknown) while its change is under way and until a look after it, never failed +// from a look at what was being replaced: on 2026-10-25 15:27:36 the controller's canary on novox was held on a have +// saying applying and failed, two seconds before its launch said healthy. +func TestAHealthIsNotJudgedWhileAModuleIsApplied(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + ctx := context.Background() + old := ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("1", 64)} + q.NoteModule(ctx, old) + q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy})) + if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthFailed { + t.Fatalf("the running module's failed look was not said: %+v", h) + } + next := ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("b", 64), Composed: "sha256:" + strings.Repeat("2", 64)} + applying := next + applying.Phase = string(delivery.PhaseApplying) + q.NoteModule(ctx, applying) + if h := l.have(t, "n1.ctl"); h.Phase != delivery.PhaseApplying || h.Health != delivery.HealthUnknown { + t.Fatalf("applying carried the last look's health: %+v", h) + } + q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy})) + if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthUnknown { + t.Fatalf("a look during the apply judged it: %+v", h) + } + q.NoteModule(ctx, next) + if h := l.have(t, "n1.ctl"); h.Phase != delivery.PhaseRunning || h.Health != delivery.HealthUnknown { + t.Fatalf("running before a look after the apply: %+v", h) + } + q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateHealthy})) + if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthOK { + t.Fatalf("the look after the apply was not taken: %+v", h) + } +} + +// A fact naming another composition than the one looked at, with no apply phase between, is not judged by the old look. +func TestAHealthIsNotCarriedOverToAnotherComposition(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + ctx := context.Background() + q.NoteModule(ctx, ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("a", 64)}) + q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy})) + q.NoteModule(ctx, ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("b", 64)}) + if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthUnknown { + t.Fatalf("the old composition's look judged the new one: %+v", h) + } +} + +// A declaration's report naming another composition of a module, with no health of its own, does not carry the old +// look's health over; and a module removed leaves nothing marked unjudged behind. +func TestADeclarationReportMovesAHealthAndARemovalForgetsIt(t *testing.T) { + l := newHaveLink() + q, _ := aHaveQueue(l) + ctx := context.Background() + q.SayFacts(ctx, Report{Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1"}}, + Health: healthOf(map[string]string{"a": StateUnhealthy})}) + if h := l.have(t, "n1.a"); h.Health != delivery.HealthFailed { + t.Fatalf("the failed look was not said: %+v", h) + } + q.SayFacts(ctx, Report{Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1", FP: "src2:" + strings.Repeat("c", 64)}}}) + if h := l.have(t, "n1.a"); h.Health != delivery.HealthUnknown { + t.Fatalf("the old composition's look judged the new one: %+v", h) + } + select { + case <-q.HealthOwed(): + default: + t.Fatal("a module out of motion and unjudged owed no look") + } + q.NoteRemoved(ctx, "a", "") + q.haves.mu.Lock() + left := q.haves.unjudged["a"] + q.haves.mu.Unlock() + if left { + t.Fatal("a removed module is still marked unjudged") + } +} diff --git a/internal/link/session.go b/internal/link/session.go index b90f733..0da7943 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -57,6 +57,8 @@ type Session struct { // gone is closed when the link open now closes, and changed whenever a link opens or closes (WhenLinked). gone chan struct{} changed chan struct{} + // owed is HealthOwed's: a module left motion unjudged. + owed chan struct{} // saying is held while a facts event is made and said, so they leave in the order they are made. saying sync.Mutex @@ -76,9 +78,27 @@ func (q *Session) init() { } q.epoch = time.Now().UnixMilli() q.changed = make(chan struct{}) + q.owed = make(chan struct{}, 1) }) } +// HealthOwed is told, at most once until it is read, when a module's change ended and no look has judged it since: the +// health loop looks and says its statement at once rather than at its next tick or five minutes on, so a module with +// nothing long-running is ok within a look of its apply, not five minutes after it (2026-10-25, review of fix 15). +func (q *Session) HealthOwed() <-chan struct{} { + q.init() + return q.owed +} + +// oweHealth nudges the health loop; a nudge already waiting is enough. +func (q *Session) oweHealth() { + q.init() + select { + case q.owed <- struct{}{}: + default: + } +} + // SayFacts says this node's facts on the link open now, in the node-engine's own order, and whether the bus took them. // A facts event made with no link open is not kept: the next link says the facts as they are then. func (q *Session) SayFacts(ctx context.Context, r Report) bool {