novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

AS15, AS16: a provider counts healthy by the node-engine seat's running, and a put ends on it

Browse Source 
The assigner hears running.<node>.<module> through its durable consumer on the seat's stream (last message per
subject), counts a provider healthy while its running build is not stopped whatever a newer want of it waits for,
and ends a put when running names the put's fingerprint (a running without one, or stopped, ends none); the have
no longer ends a put. With the stream not heard, rule 5 reads the have as before.

Builds: design 47 G §199, §309; main file "A provider before its consumers" and "Until milestone G, from the have; from G, from running"
Tests: TestAProviderRunsHealthyByItsRunning, the put rows of TestTheBusMovesByTheRulesAndItsUserListRevisionIsWrittenAtOnce, TestAnAnswerCountsOnlyForTheAskingItAnswered (running ends the put)
Manifest: consumes node-engine.running; validator: no finding new against trunk.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 12:05:11 +01:00
parent 7d1ce3296d
commit ec14823429
10 changed files with 188 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+1
View File
@@ -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)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+15 -2
View File
@@ -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)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go
+3
View File
@@ -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 {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/running.go
+88
View File
@@ -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.<node>.<module>` 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
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go
+6 -2
View File
@@ -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 `<node>.<module>`, 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 {
modules/mesh-delivery-assigner/internal/assign/put.go
+13 -19
View File
@@ -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.<node>.<module>` 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
modules/mesh-delivery-assigner/internal/assign/wants.go
+16 -1
View File
@@ -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, ""
}
}
modules/mesh-delivery-assigner/internal/assign/wants_test.go
+42 -5
View File
@@ -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)
}
}
}
modules/mesh-delivery-assigner/manifest_test.go
+2 -2
View File
@@ -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
modules/mesh-delivery-assigner/module.json
+2 -1
View File
@@ -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",