diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index d22910f4..feb6aa62 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -656,8 +656,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // composition bucket) is never applied. While the seat's assignment holds no key for this node it composes // nothing and keeps running what it runs; a assignment it cannot watch stops the agent, said, and it begins again. composer := selfComposer(session, node, hooks, say) - machinesNow.Store(composer) // the machines a failing network part points at, by the assignments (issue 774) - composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone + machinesNow.Store(composer) // the machines a failing network part points at, by the assignments (issue 774) + composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone + session.SetManifestOf(composer.ManifestOf) // the event running names the manifest of the build that runs env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does @@ -670,6 +671,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus cancel() current.set(nil) composerNow.Store(nil) + session.SetManifestOf(nil) shadowNow.Store(nil) if ctx.Err() != nil { return @@ -1076,6 +1078,8 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f // And every composition it makes in its composition state composed-, a record the provisioner, the assigner // and the view read and nothing applies (novox/hq ADR 0593 §6). env.ComposedState = link.EngineComposed{Session: session, Node: node} + // And each composition it writes there said as the seat's event composed, which the assigner waits on. + env.SayComposed = session.SayComposed return selfcompose.New(env) } diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 045e94ed..88e01c1f 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -204,11 +204,15 @@ func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string { // SayEngine says one of the seat's events about this node, on the events stream, said again through a transport fault // (sayOnce). func (l *natsLink) SayEngine(event string, body any) error { + return l.SayEngineOn(EngineEventSubject(event, l.node), body) +} + +// SayEngineOn says one of the seat's events on its subject, which names the node and, for some, the module. +func (l *natsLink) SayEngineOn(subject string, body any) error { raw, err := json.Marshal(body) if err != nil { return err } - subject := EngineEventSubject(event, l.node) return sayOnce(func(id string) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() diff --git a/internal/link/have.go b/internal/link/have.go index 063ac250..ccbf71b9 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -9,6 +9,7 @@ import ( nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" sdkkeys "git.novox.be/novox/mesh-sdk/go/keys" "github.com/novox/mesh-host/internal/compositions" + "slices" "sort" "strings" "sync" @@ -67,6 +68,11 @@ type haveState struct { userList map[string]nodeengine.UserListWritten // userListNoKey is each module whose user list was said to have no have key to ride on, so it is said once. userListNoKey map[string]bool + // runningSaid is, by module, the fingerprint and manifest its event running last said, so each is said once; and + // capsSaid the capabilities the event capabilities-changed last said (nil: none said yet this run). + runningSaid map[string]string + capsSaid []string + capsKnown bool // 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 @@ -360,6 +366,14 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { return } s := &q.haves + // What the pass says as the seat's events (running, capabilities-changed) is said once the haves' lock is let go: a + // publish never holds it. + var said []func() + defer func() { + for _, say := range said { + say() + } + }() s.mu.Lock() defer s.mu.Unlock() if s.modules == nil && s.node == nil && len(s.removed) == 0 && len(s.userList) == 0 { @@ -426,6 +440,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { s.written = map[string]string{} } s.written[module] = f.Build + if ev, ok := s.runningToSay(module, h, q.manifestOf()); ok { + said = append(said, func() { q.sayRunning(ev) }) + } } if !q.writeRemoved(write, bucket, node) { return @@ -450,6 +467,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { }) { return } + if ev, ok := s.capabilitiesToSay(node, facts.Record); ok { + said = append(said, func() { q.sayCapabilities(ev) }) + } } } if err != nil { @@ -885,3 +905,75 @@ func nothingJudged(f ModuleFact, looked bool) string { } return nodeengineclient.HealthUnknown } + +// runningToSay is the event running for a module whose have was just written running and healthy at a fingerprint whose +// manifest is known, when it has not been said for that fingerprint and manifest (design 47, *What each step says*): the +// module, the fingerprint and the digest of the manifest of that build. A module not yet healthy, or one whose manifest +// the node does not know, says nothing, and is said at the write that knows. Under s.mu. +func (s *haveState) runningToSay(module string, h nodeengine.HaveByNodeModuleValue, manifest func(module, fp string) string) (nodeengine.Running, bool) { + if h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthOK || h.Fp == "" || manifest == nil { + return nodeengine.Running{}, false + } + digest := manifest(module, h.Fp) + if digest == "" { + return nodeengine.Running{}, false + } + if s.runningSaid[module] == h.Fp+"\x00"+digest { + return nodeengine.Running{}, false + } + if s.runningSaid == nil { + s.runningSaid = map[string]string{} + } + s.runningSaid[module] = h.Fp + "\x00" + digest + return nodeengine.Running{V: sdkkeys.Version, Module: module, Fingerprint: h.Fp, Manifest: digest}, true +} + +// capabilitiesToSay is the event capabilities-changed when the node's record says other capabilities than the event last +// said (or none was said this run). Under s.mu. +func (s *haveState) capabilitiesToSay(node string, rec *nodeengine.NodeRecord) (nodeengine.CapabilitiesChanged, bool) { + if rec == nil { + return nodeengine.CapabilitiesChanged{}, false + } + if s.capsKnown && slices.Equal(s.capsSaid, rec.Capabilities) { + return nodeengine.CapabilitiesChanged{}, false + } + s.capsKnown, s.capsSaid = true, slices.Clone(rec.Capabilities) + caps := slices.Clone(rec.Capabilities) + if caps == nil { + caps = []string{} + } + return nodeengine.CapabilitiesChanged{V: sdkkeys.Version, Node: node, Capabilities: caps}, true +} + +// sayRunning says the event running, on its subject `running..`. +func (q *Session) sayRunning(ev nodeengine.Running) { + ev.Node, ev.At = q.Membership.Node, time.Now().UTC() + q.sayEngineOn(nodeengine.SubjectRunning(ev.Node, ev.Module), "running", ev) +} + +// sayCapabilities says the event capabilities-changed, on its subject `capabilities-changed.`. +func (q *Session) sayCapabilities(ev nodeengine.CapabilitiesChanged) { + ev.At = time.Now().UTC() + q.sayEngineOn(nodeengine.SubjectCapabilitiesChanged(ev.Node), "capabilities-changed", ev) +} + +// SetManifestOf gives the session what names the manifest of the build a module runs (the composer's picture of the +// catalog half): its digest, or "" when the node does not know it. +func (q *Session) SetManifestOf(f func(module, fp string) string) { + q.manifest.Lock() + q.manifestFn = f + q.manifest.Unlock() +} + +func (q *Session) manifestOf() func(module, fp string) string { + q.manifest.Lock() + defer q.manifest.Unlock() + return q.manifestFn +} + +// SayComposed says the event composed for a composition written at a fingerprint, on `composed.`. +func (q *Session) SayComposed(module, fingerprint, digest string) { + node := q.Membership.Node + q.sayEngineOn(nodeengine.SubjectComposed(node), "composed", nodeengine.Composed{V: sdkkeys.Version, Node: node, + Module: module, Fingerprint: fingerprint, Digest: digest, At: time.Now().UTC()}) +} diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 18f50d85..a575afee 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -44,6 +44,29 @@ type haveLink struct { // sayFails fails that many of the node seat's says, as a publish past the bus's retry bound; said is every event said. sayFails int said []string + // onSaid are the events said on a subject of their own (running, capabilities-changed), by subject, in order. + onSaid []string +} + +func (l *haveLink) SayEngineOn(subject string, body any) error { + raw, _ := json.Marshal(body) + l.hmu.Lock() + l.onSaid = append(l.onSaid, subject+" "+string(raw)) + l.hmu.Unlock() + return nil +} + +// saidOn is the events said on a subject, as the bus would carry them. +func (l *haveLink) saidOn(prefix string) []string { + l.hmu.Lock() + defer l.hmu.Unlock() + var out []string + for _, e := range l.onSaid { + if strings.HasPrefix(e, prefix) { + out = append(out, e) + } + } + return out } func (l *haveLink) SayEngine(event string, body any) error { @@ -326,6 +349,10 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { if l.kv["n1.node"].rev != at { t.Errorf("a report changing nothing wrote the node key again") } + // Design 47: the node's capabilities, said when they change, on capabilities-changed.. + if got := l.saidOn("mesh.seat.node-engine.event.capabilities-changed.n1 "); len(got) != 1 || !strings.Contains(got[0], `"capabilities":["apt","bus-state-calls","ledger-calls","systemd"]`) { + t.Errorf("capabilities-changed said %v, want once for the record's capabilities", got) + } // A runner that does not answer states what it stated before; one answering without any states none (an older runner). l.statedErr = fmt.Errorf("%w: nothing answered", ErrNoAnswer) l.stated = nil @@ -659,6 +686,20 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if h.Fp != fp || h.Composed != dig || h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthUnknown { t.Fatalf("the have after the apply: %+v", h) } + // Design 47, *What each step says*: running is said once, when the build is healthy and its manifest is known, on + // running.., with the module, the fingerprint and the manifest's digest; never while it is unknown. + q.SetManifestOf(func(module, f string) string { + if module == "web" && f == fp { + return "sha256:" + strings.Repeat("c", 64) + } + return "" + }) + q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) + q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) + if got := l.saidOn("mesh.seat.node-engine.event.running.n1.web "); len(got) != 1 || + !strings.Contains(got[0], `"fingerprint":"`+fp+`"`) || !strings.Contains(got[0], `"manifest":"sha256:`+strings.Repeat("c", 64)+`"`) { + t.Fatalf("running said %v, want one for the healthy build", got) + } // Starting: it waits. Past its start window the check fails: failed. q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateStarting})) if got := l.have(t, "n1.web"); got.Health != nodeengineclient.HealthWaits || got.Fp != fp || got.Composed != dig { diff --git a/internal/link/session.go b/internal/link/session.go index e4cdf90c..2d74daf2 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -103,6 +103,9 @@ type Session struct { // haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132). haves haveState + // manifestFn names the manifest of the build a module runs, for the event running (SetManifestOf), under manifest. + manifest sync.Mutex + manifestFn func(module, fp string) string // healthWritten is this node's health report as last written or read (WriteNodeHealth): the health loop alone // writes it, one look at a time. healthWritten *NodeHealth @@ -463,6 +466,20 @@ func (q *Session) SayReads() { Node: q.Membership.Node, ManifestReads: manifestReads(), At: time.Now().UTC().Format(time.RFC3339Nano), Why: why}) } +// sayEngineOn says one of the node seat's events on a subject of its own (running, composed: a module in it), over the link +// open now; with none open it is not said. +func (q *Session) sayEngineOn(subject, event string, body any) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + if sayer, ok := bus.(interface{ SayEngineOn(string, any) error }); ok { + if err := sayer.SayEngineOn(subject, body); err != nil { + q.Say("could not say the node seat's " + event + ": " + err.Error()) + } + } +} + // SayEngine says one of the node seat's events over the link open now. With none open, or a publish that failed past the // bus's retry bound, it is kept and said at the next link's opening (SayReads; novox/hq ADR 0668 §4, issue 1966): an // event a reader waits on is never only a line in this engine's log. diff --git a/internal/selfcompose/composedstate.go b/internal/selfcompose/composedstate.go index 6793a857..aef6dfbc 100644 --- a/internal/selfcompose/composedstate.go +++ b/internal/selfcompose/composedstate.go @@ -137,6 +137,7 @@ func (c *Composer) writeComposed(ctx context.Context, at *published) (wrote, del } c.pubKept[key] = ComposedEntry{Value: values[key], Revision: rev} wrote++ + c.saidComposed(key, values[key]) } for _, key := range c.pastKeeping() { if err := c.env.ComposedState.Delete(ctx, key); err != nil { @@ -319,3 +320,24 @@ func ranOf(running []engcomp.Composed) map[string]string { } return out } + +// saidComposed says the event composed for a key just written that holds a module's composition at a fingerprint and was +// not refused: the node's own composition and a refusal say nothing. +func (c *Composer) saidComposed(key string, value []byte) { + if c.env.SayComposed == nil { + return + } + _, module, fp, isNode, err := engcomp.ParseKey(key) + if err != nil || isNode { + return + } + comp, err := engcomp.Decode(value) + if err != nil || comp.Refused != "" { + return + } + digest, err := engcomp.DigestOf(comp) + if err != nil { + return + } + c.env.SayComposed(module, fp, digest) +} diff --git a/internal/selfcompose/composedstate_test.go b/internal/selfcompose/composedstate_test.go index 94a25d70..38c02b51 100644 --- a/internal/selfcompose/composedstate_test.go +++ b/internal/selfcompose/composedstate_test.go @@ -84,6 +84,14 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { state := &fakeComposed{kv: map[string]ComposedEntry{}} env := b.env() env.ComposedState = state + // Design 47, *What each step says*: each module composition written says composed, with its fingerprint and digest. + var saidMu sync.Mutex + var saidComposed []string + env.SayComposed = func(module, fingerprint, digest string) { + saidMu.Lock() + defer saidMu.Unlock() + saidComposed = append(saidComposed, module+" "+fingerprint+" "+digest) + } ctx, cancel := context.WithCancel(context.Background()) defer cancel() c := New(env) @@ -136,6 +144,14 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { t.Errorf("the published composition of lemurs: %+v, %v", comp, err) } } + if comp, err := engcomp.Decode(state.kv[lemursKey].Value); err == nil { + dig, _ := engcomp.DigestOf(comp) + saidMu.Lock() + if len(saidComposed) != 1 || saidComposed[0] != "lemurs "+fp+" "+dig || dig == "" { + t.Errorf("composed said %v, want once for lemurs at %s with its digest %q (the node's own says none)", saidComposed, fp, dig) + } + saidMu.Unlock() + } // A pass that changes nothing writes nothing. setFor(t, &askedWithin, 0) if _, err := c.ComposeNow(ctx); err != nil { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 3810cab4..1bb6fa1c 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -118,11 +118,26 @@ type Env struct { // SeatList is the node's reader of the seat list (novox/hq ADR 0636 §1): until the list was read, the composer hands // nothing new of a module whose seats are not all bootstrapped, and keeps what runs of it. Nil gates nothing. SeatList SeatGate + // SayComposed says the seat's event composed for a composition of a module written into the composition state at a + // fingerprint (design 47, *What each step says*): the module, the fingerprint and the digest of the composition. Nil says + // nothing. + SayComposed func(module, fingerprint, digest string) // SeatListChanged says the reader's held set changed: the node composes again, so what waited for the list is // handed at the read and not on a timer. Nil hears nothing. SeatListChanged <-chan struct{} } +// ManifestOf is the digest of the manifest of the build a module runs at fp, from the composer's picture of the catalog half; +// "" when the picture holds no such build (the event running names it, so one not known is not said). +func (c *Composer) ManifestOf(module, fp string) string { + c.mu.Lock() + defer c.mu.Unlock() + if m, ok := c.modules[module]; ok && m.Fp2 == fp { + return m.Manifest + } + return "" +} + // SeatGate is what the composer asks of the reader of the seat list: whether it may compose a module that claims, // invokes or provides these seats (seatlist.Reader.MayAct). type SeatGate interface {