diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ed9dab2..1af26f74 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -503,6 +503,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) + hearRunning(ctx, js, w, wake, findings, logger.Printf) } if bus != nil && kvs[buckets.Puts] != nil { p.acts = newPutActs(bus, wake, logger.Printf) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 5e8e1ce4..716ac50c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -651,8 +651,21 @@ func putOfTheBus(t *testing.T) { t.Errorf("the warrant was recorded as %q", call) } world(buckets.Wants, "want.novox.nats", wantsKV.values["want.novox.nats"]) - running, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - world(buckets.Have, hkey, running) + // Design 47 G §309: the put ends on the node-engine seat's running naming its fingerprint, never on the have. + haveRunning, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) + world(buckets.Have, hkey, haveRunning) + if got := step(); got.Step != puts.Wanted { + t.Fatalf("the have runs it and no running says so, the put stands at %s: %s", got.Step, got.Words) + } + if err := p.w.hearRunning("mesh.seat.node-engine.event.running.novox.nats", []byte(`{"v":1,"at":"2026-11-10T20:00:00Z","node":"novox","module":"nats"}`)); err != nil { + t.Fatal(err) + } + if got := step(); got.Step != puts.Wanted { + t.Fatalf("a running without a fingerprint ends none, the put stands at %s: %s", got.Step, got.Words) + } + if err := p.w.hearRunning("mesh.seat.node-engine.event.running.novox.nats", []byte(`{"v":1,"at":"2026-11-10T20:01:00Z","node":"novox","module":"nats","fingerprint":"`+fp+`"}`)); err != nil { + t.Fatal(err) + } if got := step(); got.Step != puts.Holding { t.Fatalf("running healthy, the put stands at %s: %s", got.Step, got.Words) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 77ccfaf2..0322e53c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -221,6 +221,9 @@ func (p *passer) putWorld(node, module string, put puts.Put, manifest catalogue. } } } + if r, ok := p.w.runningOf(node, module); ok { + w.Running = &r + } if hkey, err := nodeengineclient.HaveKey(node, module); err == nil { if raw := p.w.value(buckets.Have, hkey); raw != nil { if h, err := nodeengineclient.DecodeHave(raw); err == nil { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/running.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/running.go new file mode 100644 index 00000000..1aa6d079 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/running.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assignment" +) + +// What each node runs is heard, never asked (design 47 G §199, §309): the node-engine seat's event `running..` is +// kept last per subject on its own stream, and the assigner's durable consumer on it starts from the last message of every +// subject, so a node away still has its last word here. The assigner counts a provider healthy by it (rule 5) and ends a put on it. + +// hearRunning keeps each `running` the stream holds and each it says from now, and wakes the pass on every one. +func hearRunning(ctx context.Context, js jetstream.JetStream, w *world, wake func(), findings assign.Findings, logf func(string, ...any)) { + acct := process.Account{Node: thisNode, Module: assignment.Module} + nh := notHeardWriter(findings, logf) + cc, err := acct.HearSeatEvents(ctx, js, nodeengine.StreamRunning, func(m jetstream.Msg) { + if err := m.Ack(); err != nil { + logf("acknowledging %s: %v", m.Subject(), err) + } + if err := w.hearRunning(m.Subject(), m.Data()); err != nil { + logf("%v", err) + return + } + wake() + }, logf) + if err != nil { + nh.NotHeard(nodeengine.StreamRunning, "what each node runs is not heard until the assigner starts again, so a provider is counted healthy by its have and a put does not end", err) + return + } + w.mu.Lock() + w.runningHeard = true + w.mu.Unlock() + nh.Heard(nodeengine.StreamRunning) + go func() { + <-ctx.Done() + cc.Stop() + }() +} + +// hearRunning keeps one `running` event said on subject. +func (w *world) hearRunning(subject string, data []byte) error { + var r nodeengine.Running + if err := json.Unmarshal(data, &r); err != nil { + return fmt.Errorf("%s holds what is not a running: %w", subject, err) + } + if r.Node == "" || r.Module == "" { + return fmt.Errorf("%s names no node or no module, so it is kept for none", subject) + } + w.mu.Lock() + w.running[r.Node+"."+r.Module] = r + w.mu.Unlock() + return nil +} + +// runningOf is the last `running` heard of module on node. +func (w *world) runningOf(node, module string) (nodeengine.Running, bool) { + w.mu.Lock() + defer w.mu.Unlock() + r, ok := w.running[node+"."+module] + return r, ok +} + +// runningNow is every `running` heard, by node and module; nil while the seat's events are not heard (the rules then read the +// have). +func (w *world) runningNow() map[string]map[string]nodeengine.Running { + w.mu.Lock() + defer w.mu.Unlock() + if !w.runningHeard { + return nil + } + out := map[string]map[string]nodeengine.Running{} + for _, r := range w.running { + if out[r.Node] == nil { + out[r.Node] = map[string]nodeengine.Running{} + } + out[r.Node][r.Module] = r + } + return out +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index a5a1ca9a..be881eb6 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -52,6 +52,10 @@ type world struct { // sourceGroups is, by revision of the gate's source.main, the release (group) that moved it, from every revision its // history keeps (watched with history): a build's source revision names the release that carries it (ADR 0601 §1). sourceGroups map[uint64]string + // running is, by `.`, the last `running` the node-engine seat said of it, kept from its stream by the assigner's + // durable consumer (design 47 G §199, §309); runningHeard says that consumer is bound. + running map[string]nodeengine.Running + runningHeard bool } // entryMeta is a value's revision in its bucket and when it was written. @@ -70,7 +74,7 @@ type parsed struct { func newWorld() *world { return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{}, meta: map[string]entryMeta{}, trunkBuilt: map[string]builds.BuildLogEntry{}, assignmentKeys: mmac.NewKeys(), - sourceGroups: map[uint64]string{}} + sourceGroups: map[uint64]string{}, running: map[string]nodeengine.Running{}} } // putEntry is put, keeping the value's revision and the time it was written. @@ -391,7 +395,7 @@ func (w *world) wantsOf() map[string]map[string]mwc.Want { // nodes is what the six rules read of the nodes: presence, haves, and every provision's providers with their nodes. func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[string][]string) assign.Nodes { - n := assign.Nodes{Online: map[string]bool{}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}, Providers: map[string]map[string][]string{}, + n := assign.Nodes{Online: map[string]bool{}, Running: w.runningNow(), Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}, Providers: map[string]map[string][]string{}, Reads: map[string][]string{}} for key, value := range w.read(buckets.Presence) { if node, online, ok := presenceOf(key, value); ok { diff --git a/modules/mesh-delivery-assigner/internal/assign/put.go b/modules/mesh-delivery-assigner/internal/assign/put.go index 34d7ae9b..eb139832 100644 --- a/modules/mesh-delivery-assigner/internal/assign/put.go +++ b/modules/mesh-delivery-assigner/internal/assign/put.go @@ -2,7 +2,6 @@ package assign import ( "fmt" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" @@ -50,7 +49,10 @@ type PutWorld struct { SnapshotFailed string // Have is what the node says it runs of the module; nil when it says nothing. Have *nodeengine.HaveByNodeModuleValue - Now time.Time + // Running is the node-engine seat's last `running` of the module on the node; nil when it said none (design 47 G §309). A put + // ends on it, never on the have. + Running *nodeengine.Running + Now time.Time } // PutNext is what a step decides: the put as it stands after it, and the acts the pass does for it. @@ -154,26 +156,18 @@ func StepPut(w PutWorld) PutNext { return next } -// applied is a wanted put once the node's have says how it went: holding or done when it runs the composition healthy. +// applied is a wanted put once the node says how it went: holding or done when the node-engine seat's `running` names the put's +// fingerprint (design 47, *Compositions, wants and the apply*: from G a put ends on `running..` naming its +// fingerprint, never on the have). A `running` without a fingerprint, or one that says the module stopped, ends none. The have +// still says a failed health check in the put's words. func applied(w PutWorld, p puts.Put) puts.Put { - h := w.Have - if h == nil || h.Fp != p.FP { - return p - } - // A put without a hold ends on the fingerprint alone: a node that composes its own digests no composition of the put's, and - // a put that waits for that one pins the module's want for good (issue 1966). A held put keeps its composition. - if !p.Hold && h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK { + if r := w.Running; r != nil && !r.Stopped && r.Fingerprint != "" && r.Fingerprint == p.FP { + if p.Hold { + return p.Said(w.Now, puts.Holding, fmt.Sprintf("%s runs %s at %s healthy; held there until released", w.Node, w.Module, short(p.FP))) + } return p.Said(w.Now, puts.Done, fmt.Sprintf("%s runs %s at %s healthy; the rules write its want again", w.Node, w.Module, short(p.FP))) } - if h.Composed != p.Composed { - return p - } - switch { - case h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK && p.Hold: - return p.Said(w.Now, puts.Holding, fmt.Sprintf("%s runs %s at %s healthy; held there until released", w.Node, w.Module, short(p.FP))) - case h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK: - return p.Said(w.Now, puts.Done, fmt.Sprintf("%s runs %s at %s healthy; the rules write its want again", w.Node, w.Module, short(p.FP))) - case failed(*h): + if h := w.Have; h != nil && h.Fp == p.FP && h.Composed == p.Composed && failed(*h) { return p.Said(w.Now, p.Step, fmt.Sprintf("%s runs %s at %s and its health check failed: %s", w.Node, w.Module, short(p.FP), h.Detail)) } return p diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index 6e7845cd..63a15400 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -109,6 +109,10 @@ func (m Module) composition(node, fp string) (ComposedDigest, bool) { type Nodes struct { Online map[string]bool Haves map[string]map[string]nodeengine.HaveByNodeModuleValue + // Running is, by node and module, the last `running` the node-engine seat said (design 47 G §309): a provider runs healthy + // while the build it runs does, as this says it, whatever a newer want of it waits for (rule 5, main file *A provider + // before its consumers*). Nil: the seat's events are not heard, and the have says it. + Running map[string]map[string]nodeengine.Running // Providers is, by provision, the modules that provide it, with the nodes each is assigned to. Providers map[string]map[string][]string // Reads is, by node, the reader capabilities its node-engine states in its node record (nodeengine.NodeRecord.ManifestReads, @@ -276,6 +280,17 @@ func (m Module) MissingProvisions(node string) []string { return out } +// runsHealthy says whether module runs healthy on node: from G the node-engine seat's last `running` of it names a build and +// does not say it stopped; with the seat's events not heard, the have's phase and health. +func (n Nodes) runsHealthy(node, module string) bool { + if n.Running != nil { + r, ok := n.Running[node][module] + return ok && !r.Stopped && r.Fingerprint != "" + } + h, ok := n.Haves[node][module] + return ok && h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK +} + // providedHealthy is rule 5 for one provision: some module providing it runs healthy on the node, or, assigned // elsewhere only, at its canary. A provision no module provides is not a module's to judge (the mesh or the node // offers it) and holds. @@ -295,7 +310,7 @@ func providedHealthy(provision, node string, n Nodes) (bool, string) { if !slices.Contains(nodes, node) { at = Canary(nodes, n.Online) } - if h, ok := n.Haves[at][p]; ok && at != "" && h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK { + if at != "" && n.runsHealthy(at, p) { return true, "" } } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index 1c38e3a7..e3d46708 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -923,7 +923,7 @@ func stepsOfAPut(t *testing.T) { w := bus() w.Put = w.Put.Said(now, puts.Wanted, "wanted") w.Put.Composed = d.Digest - w.Have = &nodeengine.HaveByNodeModuleValue{Fp: fNew, Composed: d.Digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} + w.Running = &nodeengine.Running{Node: "novox", Module: "nats", Fingerprint: fNew} return w }, puts.Holding, "held there until released", false, false, false}, {"wanted without hold: done", func() PutWorld { @@ -931,7 +931,7 @@ func stepsOfAPut(t *testing.T) { w.Put.Hold = false w.Put = w.Put.Said(now, puts.Wanted, "wanted") w.Put.Composed = d.Digest - w.Have = &nodeengine.HaveByNodeModuleValue{Fp: fNew, Composed: d.Digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} + w.Running = &nodeengine.Running{Node: "novox", Module: "nats", Fingerprint: fNew} return w }, puts.Done, "the rules write its want again", false, false, false}, {"wanted without hold, the node composes its own: done at the fingerprint", func() PutWorld { @@ -939,14 +939,27 @@ func stepsOfAPut(t *testing.T) { w.Put.Hold = false w.Put = w.Put.Said(now, puts.Wanted, "wanted") w.Put.Composed = d.Digest - w.Have = &nodeengine.HaveByNodeModuleValue{Fp: fNew, Composed: dig(fOld), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} + w.Running = &nodeengine.Running{Node: "novox", Module: "nats", Fingerprint: fNew} return w }, puts.Done, "the rules write its want again", false, false, false}, - {"wanted with hold, the node composes its own: stays wanted", func() PutWorld { + {"wanted, the node's running names no fingerprint: stays wanted (the have runs it, which ends nothing)", func() PutWorld { w := bus() w.Put = w.Put.Said(now, puts.Wanted, "wanted") w.Put.Composed = d.Digest - w.Have = &nodeengine.HaveByNodeModuleValue{Fp: fNew, Composed: dig(fOld), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} + w.Running = &nodeengine.Running{Node: "novox", Module: "nats"} + w.Have = &nodeengine.HaveByNodeModuleValue{Fp: fNew, Composed: d.Digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} + return w + }, puts.Wanted, "wanted", false, false, false}, + {"wanted, the node's running says it stopped: stays wanted", func() PutWorld { + w := bus() + w.Put = w.Put.Said(now, puts.Wanted, "wanted") + w.Running = &nodeengine.Running{Node: "novox", Module: "nats", Fingerprint: fNew, Stopped: true} + return w + }, puts.Wanted, "wanted", false, false, false}, + {"wanted, the node's running names another build: stays wanted", func() PutWorld { + w := bus() + w.Put = w.Put.Said(now, puts.Wanted, "wanted") + w.Running = &nodeengine.Running{Node: "novox", Module: "nats", Fingerprint: fOld} return w }, puts.Wanted, "wanted", false, false, false}, {"released: nothing more", func() PutWorld { w := bus(); w.Put = w.Put.Said(now, puts.Released, "released"); return w }, puts.Released, "released", false, false, false}, @@ -1546,3 +1559,27 @@ func TestAHeldModuleWithNoWantIsNotOnNeitherList(t *testing.T) { } } } + +// Main file *A provider before its consumers* (design 47 G §309): from G a provider runs healthy while the build it runs is +// healthy, as the node-engine seat's `running` says it, whatever a newer want of it waits for (its have's phase waiting-yes). +func TestAProviderRunsHealthyByItsRunning(t *testing.T) { + waiting := nodeengine.HaveByNodeModuleValue{Fp: fOld, Phase: nodeengineclient.PhaseWaitingYes, Health: nodeengineclient.HealthOK} + for _, c := range []struct { + name string + running map[string]map[string]nodeengine.Running + ok bool + }{ + {"its new want waits for a yes, its old build runs", map[string]map[string]nodeengine.Running{"novox": {"nats": {Node: "novox", Module: "nats", Fingerprint: fOld}}}, true}, + {"its running build is stopped", map[string]map[string]nodeengine.Running{"novox": {"nats": {Node: "novox", Module: "nats", Stopped: true}}}, false}, + {"it said no running", map[string]map[string]nodeengine.Running{"novox": {}}, false}, + {"its running names no build", map[string]map[string]nodeengine.Running{"novox": {"nats": {Node: "novox", Module: "nats"}}}, false}, + {"the seat's events are not heard: the have says it, and waiting-yes is not running", nil, false}, + } { + n := Nodes{Online: map[string]bool{"novox": true}, Running: c.running, + Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{"novox": {"nats": waiting}}, + Providers: map[string]map[string][]string{"bus-state": {"nats": {"novox"}}}} + if ok, words := providedHealthy("bus-state", "novox", n); ok != c.ok { + t.Errorf("%s: healthy = %v (%s), want %v", c.name, ok, words, c.ok) + } + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 1db630c4..ba1b4796 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -123,8 +123,8 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { - t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) + if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { + t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index d6252329..e69ae235 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -48,7 +48,8 @@ "mesh-delivery-teller.ask-ended", "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", - "node-engine.manifest-reads-changed" + "node-engine.manifest-reads-changed", + "node-engine.running" ], "invokes": [ "mesh-delivery-indexer.graph",