From 7623279efe83b487be802554f335f0ddff33070e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:14:46 +0100 Subject: [PATCH 1/5] =?UTF-8?q?Node-engine:=20no=20compiled=20seat=20list;?= =?UTF-8?q?=20the=20seat's=20name=20and=20verbs=20come=20from=20its=20cont?= =?UTF-8?q?ract,=20the=20tests=20give=20catalogue=20their=20seats=20(H-7,?= =?UTF-8?q?=20ADR=200635=20=C2=A71;=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../frombustest/seats_for_tests_test.go | 10 ++++++++++ .../compose/frombus/seats_for_tests_test.go | 10 ++++++++++ internal/compose/resolution/files_test.go | 11 ++++++----- internal/compose/resolution/seats.go | 10 ---------- .../resolution/seats_for_tests_test.go | 10 ++++++++++ internal/compose/resolution/seats_test.go | 19 ------------------- internal/compose/seats_for_tests_test.go | 10 ++++++++++ internal/composer/seats_for_tests_test.go | 10 ++++++++++ internal/link/engineseat.go | 5 +++-- 9 files changed, 59 insertions(+), 36 deletions(-) create mode 100644 internal/compose/frombus/frombustest/seats_for_tests_test.go create mode 100644 internal/compose/frombus/seats_for_tests_test.go create mode 100644 internal/compose/resolution/seats_for_tests_test.go create mode 100644 internal/compose/seats_for_tests_test.go create mode 100644 internal/composer/seats_for_tests_test.go diff --git a/internal/compose/frombus/frombustest/seats_for_tests_test.go b/internal/compose/frombus/frombustest/seats_for_tests_test.go new file mode 100644 index 0000000..2700f7c --- /dev/null +++ b/internal/compose/frombus/frombustest/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package frombustest + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests judge against are given to catalogue as the node-engine gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/internal/compose/frombus/seats_for_tests_test.go b/internal/compose/frombus/seats_for_tests_test.go new file mode 100644 index 0000000..16c5987 --- /dev/null +++ b/internal/compose/frombus/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package frombus + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests judge against are given to catalogue as the node-engine gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/internal/compose/resolution/files_test.go b/internal/compose/resolution/files_test.go index 4d8ed14..625f40d 100644 --- a/internal/compose/resolution/files_test.go +++ b/internal/compose/resolution/files_test.go @@ -42,12 +42,13 @@ func TestTheHolderOfNodeFilesIsToldSecretPlacesAndOwnedPaths(t *testing.T) { Resources: []map[string]any{{"id": "stub", "type": "file", "path": mcat.ResolverFile}}} uplink := mcat.Manifest{Module: "networkmanager", Version: "1", Claims: []mcat.Claim{{Name: "node-uplink"}}, Resources: []map[string]any{{"id": "resolv", "type": "file", "path": mcat.ResolverFile}}} - compiled := seats - t.Cleanup(func() { seats = compiled }) + compiled := mcat.Seats() + t.Cleanup(func() { mcat.UseSeats(compiled) }) for _, order := range []string{"compiled", "reversed"} { if order == "reversed" { - seats = append([]mcat.Seat(nil), compiled...) - slices.Reverse(seats) + reversed := append([]mcat.Seat(nil), compiled...) + slices.Reverse(reversed) + mcat.UseSeats(reversed) } both := mcat.Resolution{Node: "anchor", Modules: []mcat.Manifest{files, uplink, resolver}} w, err := resolutionFilesWords(both, files, nil, mcat.Rendering{}) @@ -59,7 +60,7 @@ func TestTheHolderOfNodeFilesIsToldSecretPlacesAndOwnedPaths(t *testing.T) { t.Errorf("seat table %s: %s is owned by %q, want systemd-resolved (%s)", order, mcat.ResolverFile, o[mcat.ResolverFile], mcat.ResolverSeat) } } - seats = compiled + mcat.UseSeats(compiled) // Issue 1126, C5 and seed-8: a module carried as it runs (compose.Carried) owns its manifest's paths (OwnsPaths), as // the controller tells them, and not the paths its running resources are at (a directory the mesh placed). carried := mcat.Manifest{Module: "de-spiegel", Version: "1", OwnsPaths: []string{"/var/lib/de-spiegel/server.env"}, diff --git a/internal/compose/resolution/seats.go b/internal/compose/resolution/seats.go index 1420499..8813719 100644 --- a/internal/compose/resolution/seats.go +++ b/internal/compose/resolution/seats.go @@ -4,16 +4,6 @@ import ( mcat "git.novox.be/novox/mesh-sdk/go/catalogue" ) -// defaultSeats is the set the mesh ships with — the seed for the control plane's seat table and the fallback when it has none -// (novox/hq ADR 0122). It is mesh-contracts' generated list, as mesh-sdk compiles it, taken whole: the node-engine keeps no copy of -// a seat's verbs or events (design 26, ADR 0608 §5; issue 1966 ENG-3). -var defaultSeats = mcat.DefaultSeats() - -// seats is the working set the lookups read. It starts as the compiled defaults and is replaced by -// what the control plane loaded from its store (novox/hq ADR 0122), so a change to the set is a -// change to data, not to this code. -var seats = defaultSeats - // configInput is the schema of mesh-config's verb: a value of the key's own type (a number for a number, never its // text), and a limit that is a number; every other argument a string. func configInput() map[string]any { diff --git a/internal/compose/resolution/seats_for_tests_test.go b/internal/compose/resolution/seats_for_tests_test.go new file mode 100644 index 0000000..ef33ada --- /dev/null +++ b/internal/compose/resolution/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package resolution + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests judge against are given to catalogue as the node-engine gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/internal/compose/resolution/seats_test.go b/internal/compose/resolution/seats_test.go index bcbac2c..4a72341 100644 --- a/internal/compose/resolution/seats_test.go +++ b/internal/compose/resolution/seats_test.go @@ -1,7 +1,6 @@ package resolution import ( - "reflect" "strings" "testing" @@ -71,22 +70,4 @@ func TestTheHolderIsTheModuleNotTheMachine(t *testing.T) { t.Fatalf("the holder was not told apart from a neighbour: %+v", holder) } - // ADR 0604 section 1, F3: the seat table written here answers as the SDK's does for the catalog seat, its four - // verbs included, so no two copies of the table disagree. - want, ok := mcat.SeatNamed(mcat.CatalogSeat) - if !ok { - t.Fatal("the SDK holds no catalog seat") - } - var have *mcat.Seat - for i := range defaultSeats { - if defaultSeats[i].Name == mcat.CatalogSeat { - have = &defaultSeats[i] - } - } - if have == nil { - t.Fatal("the table here holds no catalog seat") - } - if !reflect.DeepEqual(*have, want) { - t.Errorf("the table here has %+v, the SDK's is %+v", *have, want) - } } diff --git a/internal/compose/seats_for_tests_test.go b/internal/compose/seats_for_tests_test.go new file mode 100644 index 0000000..c6afc26 --- /dev/null +++ b/internal/compose/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package compose + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests judge against are given to catalogue as the node-engine gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/internal/composer/seats_for_tests_test.go b/internal/composer/seats_for_tests_test.go new file mode 100644 index 0000000..5415dc6 --- /dev/null +++ b/internal/composer/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package composer + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests judge against are given to catalogue as the node-engine gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 5256348..4d47ace 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -25,7 +26,7 @@ import ( // tool's (mesh-tools node-tools). One service, `node-engine`, answers discovery with each verb's seat, tool and node. // EngineSeat is the seat. -const EngineSeat = catalogue.EngineSeat +const EngineSeat = nodeengine.Seat // EngineToolSubject is where a verb of this node's seat is asked naming no caller. func EngineToolSubject(verb, node string) string { @@ -117,7 +118,7 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { return } apart := make(chan struct{}, apartAtOnce) - for _, v := range catalogue.EngineVerbs() { + for _, v := range catalogue.EngineVerbs(EngineSeat) { verb := v.Name answer := EngineAnswer(handle, verb, l.node, apart) meta := EngineEndpointMetadata(v, l.node) From 66fe7c8a0ded2c39f27fb2fb26ffd04e3414950e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:16:04 +0100 Subject: [PATCH 2/5] =?UTF-8?q?Node-engine:=20the=20finding=20of=20a=20sea?= =?UTF-8?q?t=20list=20unread=20for=20five=20minutes,=20written=20once=20an?= =?UTF-8?q?d=20cleared=20at=20the=20first=20full=20read=20(H-7,=20ADR=2006?= =?UTF-8?q?36=20=C2=A71;=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/seatwait/seatwait.go | 118 +++++++++++++++++++++++ internal/seatwait/seatwait_test.go | 149 +++++++++++++++++++++++++++++ 2 files changed, 267 insertions(+) create mode 100644 internal/seatwait/seatwait.go create mode 100644 internal/seatwait/seatwait_test.go diff --git a/internal/seatwait/seatwait.go b/internal/seatwait/seatwait.go new file mode 100644 index 0000000..9c74565 --- /dev/null +++ b/internal/seatwait/seatwait.go @@ -0,0 +1,118 @@ +// Package seatwait is the node-engine's word that its reader of the seat list has waited (novox/hq ADR 0636 §1): a node +// whose reader holds only the bootstrapped definitions for five minutes writes one finding in the engine's own state +// `findings`, which the condition keeper raises as a condition and clears at the first full read (ADR 0564 §2). +// +// Event-driven: the reader says each change (Changed), and the one bound is a timer started when the wait begins and +// stopped when it ends, never a loop. A timer that fires looks at the reader again before it writes, so a read that +// ended the wait without a change of the held set (the seat answered the list the reader already held) writes nothing. +package seatwait + +import ( + "fmt" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// Bound is how long the reader may wait before the node says it (ADR 0636 §1). +const Bound = 5 * time.Minute + +// Kind is the finding's kind; its id is the node's. +const Kind = "seat-list-unread" + +// Reader is what the Waiter asks of the seat list's reader. +type Reader interface { + Status() seatlist.Status +} + +// Timer is a started timer that can be stopped. +type Timer interface{ Stop() bool } + +// Clock gives the time and starts a timer; the tests give their own, so nothing sleeps. +type Clock interface { + Now() time.Time + AfterFunc(d time.Duration, f func()) Timer +} + +// Waiter writes and clears the node's finding. +type Waiter struct { + Node string + Observer string + Reader Reader + Findings views.FindingWriter + Clock Clock + // Say tells the node's journal; nil says nothing. + Say func(string) + + mu sync.Mutex + timer Timer + since time.Time + written *views.Finding +} + +func waiting(st seatlist.Status) bool { return !st.Heard || len(st.Waiting) > 0 } + +// Changed is told when the reader's held set or its status moved: it starts the bound when the reader waits and has no +// bound running, stops it and clears the finding when the reader has read the whole list. +func (w *Waiter) Changed() { + st := w.Reader.Status() + w.mu.Lock() + defer w.mu.Unlock() + if !waiting(st) { + if w.timer != nil { + w.timer.Stop() + w.timer = nil + } + w.since = time.Time{} + if w.written != nil { + if err := w.Findings.Delete(*w.written); err != nil { + w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be cleared: %v", err)) + return + } + w.written = nil + } + return + } + if w.timer != nil || w.written != nil { + return + } + w.since = w.Clock.Now() + w.timer = w.Clock.AfterFunc(Bound, w.expired) +} + +// Waiting says whether the bound is running or the finding stands, and since when. +func (w *Waiter) Waiting() (bool, time.Time) { + w.mu.Lock() + defer w.mu.Unlock() + return w.timer != nil || w.written != nil, w.since +} + +func (w *Waiter) expired() { + st := w.Reader.Status() + w.mu.Lock() + defer w.mu.Unlock() + w.timer = nil + if !waiting(st) { + w.since = time.Time{} + return + } + f := views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node, + Summary: fmt.Sprintf("%s has read no complete seat list since %s: %s", w.Node, w.since.UTC().Format(time.RFC3339), st.Health()), + Headline: fmt.Sprintf("%s waits for the seat list", w.Node), + Explanation: "The node holds only the definitions it starts with, so it keeps running what it runs and starts, composes and grants nothing new. It reads the list again at the next change the seat mesh-catalog says, or when asked to.", + At: w.Clock.Now()} + if err := w.Findings.Put(f); err != nil { + w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be written: %v", err)) + return + } + w.written = &f + w.say(f.Summary) +} + +func (w *Waiter) say(s string) { + if w.Say != nil { + w.Say(s) + } +} diff --git a/internal/seatwait/seatwait_test.go b/internal/seatwait/seatwait_test.go new file mode 100644 index 0000000..c04db1a --- /dev/null +++ b/internal/seatwait/seatwait_test.go @@ -0,0 +1,149 @@ +package seatwait + +import ( + "errors" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" +) + +type fakeReader struct{ st seatlist.Status } + +func (f *fakeReader) Status() seatlist.Status { return f.st } + +type fakeTimer struct { + c *fakeClock + due time.Time + f func() + stopped bool +} + +func (t *fakeTimer) Stop() bool { was := !t.stopped; t.stopped = true; return was } + +type fakeClock struct { + now time.Time + timers []*fakeTimer +} + +func (c *fakeClock) Now() time.Time { return c.now } +func (c *fakeClock) AfterFunc(d time.Duration, f func()) Timer { + t := &fakeTimer{c: c, due: c.now.Add(d), f: f} + c.timers = append(c.timers, t) + return t +} + +// advance moves the clock and fires the timers that fell due and were not stopped. +func (c *fakeClock) advance(d time.Duration) { + c.now = c.now.Add(d) + for _, t := range c.timers { + if !t.stopped && !t.due.After(c.now) { + t.stopped = true + t.f() + } + } +} + +type fakeFindings struct { + put, deleted []views.Finding + err error +} + +func (f *fakeFindings) Put(x views.Finding) error { + if f.err != nil { + return f.err + } + f.put = append(f.put, x) + return nil +} +func (f *fakeFindings) Delete(x views.Finding) error { f.deleted = append(f.deleted, x); return nil } + +// ADR 0636 §1: a reader that waited five minutes makes the node write one finding, which the first full read clears. +func caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead(t *testing.T) { + reader := &fakeReader{st: seatlist.Status{Heard: false, LastError: "no answer"}} + clock := &fakeClock{now: time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC)} + findings := &fakeFindings{} + var said []string + w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Findings: findings, Clock: clock, Say: func(s string) { said = append(said, s) }} + + w.Changed() + clock.advance(Bound - time.Second) + if len(findings.put) != 0 { + t.Fatalf("a finding was written before the bound: %+v", findings.put) + } + w.Changed() // another change while waiting starts no second bound + clock.advance(time.Second) + if len(findings.put) != 1 || findings.put[0].Kind != Kind || findings.put[0].ID != "ace" { + t.Fatalf("after five minutes the node wrote %+v, want one %s finding for ace", findings.put, Kind) + } + clock.advance(time.Hour) + if len(findings.put) != 1 { + t.Fatalf("the finding was written again: %d", len(findings.put)) + } + + reader.st = seatlist.Status{Heard: true} + w.Changed() + if len(findings.deleted) != 1 || findings.deleted[0].ID != "ace" { + t.Fatalf("the first full read left the finding: deleted %+v", findings.deleted) + } + if on, _ := w.Waiting(); on { + t.Error("the waiter still waits after the full read") + } +} + +// A read that ends the wait stops the bound; a timer that fires after the list was read writes nothing. +func caseAReadWithinTheBoundWritesNothing(t *testing.T) { + reader := &fakeReader{st: seatlist.Status{Heard: false}} + clock := &fakeClock{now: time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC)} + findings := &fakeFindings{} + w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Findings: findings, Clock: clock} + w.Changed() + clock.advance(time.Minute) + reader.st = seatlist.Status{Heard: true} + w.Changed() + clock.advance(time.Hour) + if len(findings.put) != 0 || len(findings.deleted) != 0 { + t.Fatalf("a list read in time wrote %+v and cleared %+v", findings.put, findings.deleted) + } + + // The seat answered the list already held: no change was said, so the timer is the one to look again. + reader.st = seatlist.Status{Heard: false} + w.Changed() + reader.st = seatlist.Status{Heard: true} + clock.advance(Bound) + if len(findings.put) != 0 { + t.Fatalf("a timer that found the list read wrote %+v", findings.put) + } +} + +// A finding that cannot be written is said and tried again at the next change, not lost. +func caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain(t *testing.T) { + reader := &fakeReader{st: seatlist.Status{Heard: false}} + clock := &fakeClock{now: time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC)} + findings := &fakeFindings{err: errors.New("bus down")} + var said []string + w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Findings: findings, Clock: clock, Say: func(s string) { said = append(said, s) }} + w.Changed() + clock.advance(Bound) + if len(said) != 1 || len(findings.put) != 0 { + t.Fatalf("said %v, wrote %v", said, findings.put) + } + findings.err = nil + w.Changed() + clock.advance(Bound) + if len(findings.put) != 1 { + t.Fatalf("the finding was not written at the next bound: %v", findings.put) + } +} + +// ADR 0636 §1, the node's finding of a seat list unread. +func TestTheSeatListWait(t *testing.T) { + for name, run := range map[string]func(*testing.T){ + "five minutes write one finding and the first full read clears it": caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead, + "a read within the bound writes nothing": caseAReadWithinTheBoundWritesNothing, + "a finding that cannot be written is said and tried again": caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain, + } { + t.Run(name, run) + } +} From a222db8c69ebcb667c6a8507ab5ea29457b01732 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:17:28 +0100 Subject: [PATCH 3/5] =?UTF-8?q?Node-engine=20composer:=20a=20module=20new?= =?UTF-8?q?=20to=20the=20node=20is=20held=20until=20the=20seat=20list=20wa?= =?UTF-8?q?s=20read,=20unless=20its=20seats=20are=20bootstrapped;=20the=20?= =?UTF-8?q?read=20hands=20it=20(H-7,=20ADR=200636=20=C2=A71;=20issue=20196?= =?UTF-8?q?6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/selfcompose/selfcompose.go | 109 +++++++++++++++++++++++ internal/selfcompose/selfcompose_test.go | 81 +++++++++++++++++ 2 files changed, 190 insertions(+) diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 6607fc6..0bc5d21 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -102,6 +102,18 @@ type Env struct { ComposedState ComposedStore // Now is the clock what the composer waits on is timed by; nil is time.Now. Now func() time.Time + // 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 + // 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{} +} + +// 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 { + MayAct(seats ...string) bool } // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value. @@ -439,6 +451,25 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er } }() } + if c.env.SeatListChanged != nil { + go func() { + for { + select { + case <-ctx.Done(): + return + case _, ok := <-c.env.SeatListChanged: + if !ok { + return + } + select { + case merged <- tagged{in: -1, ok: true}: + case <-ctx.Done(): + return + } + } + } + }() + } out := make(chan agent.Update, 64) c.watching.Store(true) go func() { @@ -908,6 +939,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { done() return nil } + // A reader that has not read the seat list acts on nothing new (ADR 0636 §1): a module whose composition differs from + // what runs, and whose seats are not all bootstrapped, is held with what it runs. + held = c.holdUntilListed(ctx, r, compositions, held, running) // The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs. missed = c.missed.Load() for key, comp := range compositions { @@ -1731,3 +1765,78 @@ func mapValues(m map[string]delivery.Composed) []delivery.Composed { } return out } + +// WaitsForTheSeatList is why a module is held while the node has not read the seat list. +const WaitsForTheSeatList = "the node has not read the seat list, so it composes nothing new of this module and keeps what runs of it (novox/hq ADR 0636 §1)" + +// holdUntilListed takes out of compositions every module's composition that is new (no running composition has its +// digest) while the reader may not act on its seats, and names the module in held: it keeps running what it runs, and +// the pass after the list was read hands it. The node's own composition and a module whose composition runs already are +// left as they are. +func (c *Composer) holdUntilListed(ctx context.Context, r reader, compositions map[string]delivery.Composed, held map[string]string, running []delivery.Composed) map[string]string { + if c.env.SeatList == nil { + return held + } + ran := map[string]bool{} + for _, comp := range running { + if dig, err := delivery.ComposedDigestOf(comp); err == nil { + ran[comp.Module+"/"+dig] = true + } + } + for key, comp := range compositions { + if comp.Module == "" { + continue + } + if dig, err := delivery.ComposedDigestOf(comp); err == nil && ran[comp.Module+"/"+dig] { + continue + } + if c.env.SeatList.MayAct(c.seatsOf(ctx, r, comp.Module)...) { + continue + } + delete(compositions, key) + if held == nil { + held = map[string]string{} + } + if _, already := held[comp.Module]; !already { + held[comp.Module] = WaitsForTheSeatList + } + } + return held +} + +// seatsOf is the seats a module claims, invokes or provides, from the manifest its module record names; none when it +// cannot be read, which the reader judges as a module no bootstrapped seat names. +func (c *Composer) seatsOf(ctx context.Context, r reader, module string) []string { + c.mu.Lock() + m, ok := c.modules[module] + c.mu.Unlock() + if !ok || m.Manifest == "" { + return nil + } + raw, err := r.Manifest(ctx, m.Manifest) + if err != nil { + return nil + } + var man struct { + Claims []struct { + Name string `json:"name"` + } `json:"claims"` + Invokes []string `json:"invokes"` + Provides []struct { + Name string `json:"name"` + } `json:"provides"` + } + if json.Unmarshal(raw, &man) != nil { + return nil + } + var seats []string + for _, cl := range man.Claims { + seats = append(seats, cl.Name) + } + for _, in := range man.Invokes { + if seat, _, ok := strings.Cut(strings.TrimPrefix(in, "seat:"), "."); ok && strings.HasPrefix(in, "seat:") { + seats = append(seats, seat) + } + } + return seats +} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 170a0b3..542667a 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -1750,3 +1750,84 @@ func setFor[T any](t *testing.T, p *T, v T) { *p = v t.Cleanup(func() { *p = old }) } + +type gate struct{ open atomic.Bool } + +func (g *gate) MayAct(...string) bool { return g.open.Load() } + +// ADR 0636 §1: a node that has not read the seat list composes nothing new of a module whose seats are not bootstrapped, +// says it holds the module (so its have says why) and hands it at the read, which the reader says as an event; what +// already runs with the same composition is not held. +func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) { + b := newFakeBus() + fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) + digest := delivery.ManifestDigest([]byte(lemurs)) + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs}) + key, _ := delivery.ManifestsKey(digest) + b.kv[delivery.ManifestsBucket+"/"+key] = man + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + env := b.env() + g := &gate{} + changed := make(chan struct{}, 1) + env.SeatList, env.SeatListChanged = g, changed + var mu sync.Mutex + heldBy := map[string]string{} + env.Refused = func(_ string, held map[string]string) { mu.Lock(); heldBy = held; mu.Unlock() } + c := New(env) + out, err := c.Watch(ctx, "n1.>") + if err != nil { + t.Fatal(err) + } + place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []delivery.PlacedModule{{Module: "lemurs"}}}) + have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) + want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true}) + nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true}) + b.put(t, delivery.PlacementBucket, "placement.n1", place) + b.put(t, delivery.HaveBucket, "n1.node", have) + b.put(t, delivery.ModulesBucket, "lemurs", mod) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want) + b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant) + for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} { + b.send(bucket, agent.Update{Caught: true}) + } + handed := map[string]bool{} + for { + u := next(t, out) + if u.Caught { + break + } + handed[u.Key] = true + } + mu.Lock() + why := heldBy["lemurs"] + mu.Unlock() + if why != WaitsForTheSeatList { + t.Errorf("lemurs is held with %q, want %q", why, WaitsForTheSeatList) + } + for k := range handed { + if strings.Contains(k, "lemurs") { + t.Errorf("the node handed %s before it read the seat list", k) + } + } + if !handed["n1.node"] { + t.Errorf("the node's own composition was held with the module: %v", handed) + } + // The reader reads the list and says so: the module is handed, with no timer. + g.open.Store(true) + changed <- struct{}{} + for { + u := next(t, out) + if strings.Contains(u.Key, "lemurs") && !u.Deleted { + break + } + } + mu.Lock() + still := heldBy["lemurs"] + mu.Unlock() + if still != "" { + t.Errorf("lemurs is still held after the read: %q", still) + } +} From 855f4471e276a7ee3c2eeb6879cd53d798c32a52 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:26:15 +0100 Subject: [PATCH 4/5] =?UTF-8?q?Node-engine:=20reads=20the=20seat=20list=20?= =?UTF-8?q?through=20the=20seat=20mesh-catalog,=20holds=20what=20is=20new?= =?UTF-8?q?=20until=20it=20was=20read,=20writes=20its=20finding=20after=20?= =?UTF-8?q?five=20minutes;=20verbs=20seat-list=20and=20seat-list-again;=20?= =?UTF-8?q?whole-module=20case=20against=20a=20stand-in=20of=20the=20seat?= =?UTF-8?q?=20(H-7,=20ADR=200636=20=C2=A71,=20=C2=A77,=20=C2=A78;=20issue?= =?UTF-8?q?=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/main.go | 1 + cmd/mesh-host/nodeagent.go | 17 +- cmd/mesh-host/nodeagent_integration_test.go | 144 +++++++++++++++++ cmd/mesh-host/seatlist.go | 166 ++++++++++++++++++++ go.mod | 2 + internal/link/findings.go | 63 ++++++++ internal/link/handacts.go | 18 +++ internal/link/movesecret.go | 26 ++- internal/nodegrant/testdata/node.json | 7 +- 9 files changed, 436 insertions(+), 8 deletions(-) create mode 100644 cmd/mesh-host/seatlist.go create mode 100644 internal/link/findings.go diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 0d03f6b..a6d13b3 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1017,6 +1017,7 @@ func runLink(ctx context.Context, opts options) error { queue.Acts.Composition = compositionFrom(current) queue.Acts.Compose = composeFrom(current, &composerNow) queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow) + queue.Acts.SeatList, queue.Acts.SeatListAgain = seatListRead, seatListAgain queue.Acts.ComposeCandidate = composeCandidateFrom(mine.Node, &shadowNow) setRunsHere(runsFromAgent(current)) factsOfNow := func(ctx context.Context, ag *agent.Agent) { diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 57dfbde..66af382 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -586,7 +586,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in // flight at the switch marked the next agent's modules failed). own := &agentNow{} + // And the seat list it reads through the seat mesh-catalog, from its bootstrapped definitions until it has: the + // composer hands nothing new meanwhile, and the node says it has waited five minutes (novox/hq ADR 0636 §1). + seats, err := startSeatList(run, session, node, say) + if err != nil { + say(err.Error()) + } hooks := composerHooks{ + seatList: seats, running: func() []delivery.Composed { if now := own.get(); now != nil { return runsAsWritten(now.Everything(), opts.state) @@ -628,7 +635,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into // its own key, `.node`, under candidates. shadowHooks := hooks - shadowHooks.refused, shadowHooks.kept, shadowHooks.certify = nil, nil, nil + shadowHooks.refused, shadowHooks.kept, shadowHooks.certify, shadowHooks.seatList = nil, nil, nil, nil shadowEnv := composerEnv(session, node, say, shadowHooks) shadow := selfcompose.NewShadow(shadowEnv, func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { @@ -982,6 +989,8 @@ type composerHooks struct { // vaultAuthority and certify are selfcompose.Env's VaultAuthority and Certify. vaultAuthority func() string certify func(ctx context.Context) error + // seatList is this run's reader of the seat list: the composer hands nothing new until it was read (ADR 0636 §1). + seatList *seatListState } // selfComposer is the node's own composer: it composes what the seat's placement places on the node, and nothing while @@ -1003,7 +1012,7 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { - return selfcompose.Env{Node: node, + env := selfcompose.Env{Node: node, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket { // The seat's states are made when their holder starts: a node-engine that starts first waits for the @@ -1033,4 +1042,8 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com Fetch: apply.Fetch, Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held, Kept: hooks.kept, VaultAuthority: hooks.vaultAuthority, Certify: hooks.certify} + if hooks.seatList != nil { + env.SeatList, env.SeatListChanged = hooks.seatList.reader, hooks.seatList.changed + } + return env } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 4ae58f7..7eb33ea 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -18,12 +19,14 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/seatlist" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/nodegrant" + "github.com/novox/mesh-host/internal/seatwait" "github.com/novox/mesh-host/internal/selfcompose" "github.com/novox/mesh-host/internal/store" "github.com/novox/mesh-host/internal/system" @@ -55,6 +58,9 @@ type itBed struct { // refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met. refusals []string mu sync.Mutex + // seatSilent makes the stand-in of the seat mesh-catalog refuse the seat list (it is served from the start otherwise). + seatSilent atomic.Bool + findings nats.KeyValue } // aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what @@ -133,6 +139,10 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed { } *c.into = kv } + if b.findings, err = b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: link.FindingsBucket, History: 1}); err != nil { + t.Fatal(err) + } + b.standInSeat() b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second} if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil { t.Fatal(err) @@ -875,3 +885,137 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test runtime.containers[server] == made, runtime.moved(server)) } } + +// standInSeat is the seat mesh-catalog as the node reads the seat list from it (novox/hq ADR 0636 §4): with no seat, every +// bootstrapped seat with a digest; with one, its definition. While seatSilent it refuses. +func (b *itBed) standInSeat() { + defs := map[string]json.RawMessage{} + var entries []seatlist.Entry + for _, d := range seatlist.Bootstrapped() { + raw, err := json.Marshal(d) + if err != nil { + b.t.Fatal(err) + } + defs[d.Seat] = raw + entries = append(entries, seatlist.Entry{Seat: d.Seat, Source: seatlist.SourceTrunk, Digest: "d-" + d.Seat}) + } + sub, err := b.conn.Subscribe("mesh.seat."+seatlist.Seat+".call."+seatlist.Verb+".*", func(m *nats.Msg) { + var q struct { + Seat string `json:"seat"` + } + _ = json.Unmarshal(m.Data, &q) + var result any + switch { + case b.seatSilent.Load(): + _ = m.Respond([]byte(`{"error":"the indexer has not started"}`)) + return + case q.Seat == "": + result = seatlist.Listing{Revision: 1, Seats: entries} + default: + result = seatlist.Answer{Seat: q.Seat, Found: true, Revision: 1, + Entry: &seatlist.Entry{Seat: q.Seat, Source: seatlist.SourceTrunk, Digest: "d-" + q.Seat, Definition: defs[q.Seat]}} + } + out, _ := json.Marshal(map[string]any{"result": result}) + _ = m.Respond(out) + }) + if err != nil { + b.t.Fatal(err) + } + b.t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +// fiveMinutes is a clock of the test: the bound of the node's wait fires when the test says so, nothing sleeps. +type fiveMinutes struct { + mu sync.Mutex + timers []*time.Timer + fire []func() +} + +func (c *fiveMinutes) Now() time.Time { return time.Now() } +func (c *fiveMinutes) AfterFunc(_ time.Duration, f func()) seatwait.Timer { + c.mu.Lock() + defer c.mu.Unlock() + t := &stoppable{} + c.fire = append(c.fire, func() { + if !t.stopped.Load() { + f() + } + }) + return t +} + +type stoppable struct{ stopped atomic.Bool } + +func (s *stoppable) Stop() bool { return !s.stopped.Swap(true) } + +func (c *fiveMinutes) pass() { + c.mu.Lock() + fire := c.fire + c.fire = nil + c.mu.Unlock() + for _, f := range fire { + f() + } +} + +// ADR 0636 §1, §8, whole against a real bus and the stand-in of the seat: while the seat refuses, the node composes and +// applies nothing new, and keeps running what it ran; after the bound it writes one finding in its state findings; when +// the seat names its list the module is applied and the finding is cleared; a node started again with the seat gone +// applies the composition it applied last. +func TestNatsANodeThatCannotReadTheSeatListActsOnNothingNew(t *testing.T) { + clock := &fiveMinutes{} + was := seatWaitClock + seatWaitClock = clock + t.Cleanup(func() { seatWaitClock = was }) + + b := newItBedAs(t, nodegrant.For(itNode, "it2.mailer.domain")) + b.seatSilent.Store(true) + stop, st := h8(t, b, false) + defer stop() + _ = st + finding := "machine." + itNode + ".seat-list-unread" + within(t, 5*time.Second, "the node did not start its bound while the seat refused", func() bool { + clock.mu.Lock() + defer clock.mu.Unlock() + return len(clock.fire) > 0 + }) + if b.file("web") != "" { + t.Fatalf("the node applied web before it read the seat list: %q", b.file("web")) + } + if _, err := b.findings.Get(finding); err == nil { + t.Fatal("the node wrote its finding before the bound") + } + clock.pass() + within(t, 5*time.Second, "the node did not write its finding after the bound", func() bool { + _, err := b.findings.Get(finding) + return err == nil + }) + if b.file("web") != "" { + t.Fatalf("the node applied web with its finding written: %q", b.file("web")) + } + + b.seatSilent.Store(false) + // The hand act of the operator and the healers (the verb seat-list-again): a read now, as the seat's event would make. + if out, err := seatListAgain(); err != nil || out.(map[string]any)["error"] != nil { + t.Fatalf("seat-list-again: %v, %v", out, err) + } + within(t, 50*time.Second, "the node did not apply web once the seat named its list", func() bool { return b.file("web") == "greeting=bonjour\n" }) + within(t, 5*time.Second, "the node did not clear its finding at the first full read", func() bool { + _, err := b.findings.Get(finding) + return errors.Is(err, nats.ErrKeyNotFound) + }) + + // Started again with the seat refusing: nothing new is composed, and what ran is neither removed nor changed. + stop() + b.seatSilent.Store(true) + stopAgain := b.start() + defer stopAgain() + within(t, 30*time.Second, "the node did not hold web as it was after a start with the seat gone", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return strings.Contains(strings.Join(b.said, "\n"), "it1 has read no complete seat list") || b.file("web") != "" + }) + if b.file("web") != "greeting=bonjour\n" { + t.Errorf("web does not stay as it ran after a start with the seat gone: %q", b.file("web")) + } +} diff --git a/cmd/mesh-host/seatlist.go b/cmd/mesh-host/seatlist.go new file mode 100644 index 0000000..ca6daec --- /dev/null +++ b/cmd/mesh-host/seatlist.go @@ -0,0 +1,166 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync/atomic" + "time" + + conditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/seatwait" +) + +// The node-engine reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636): through the seat mesh-catalog, +// from the bootstrapped definitions until it has been read, acting on nothing new meanwhile (the composer's gate), and +// following the seat's events. What it holds is given to catalogue at every change, which compiles no seat list. + +// seatListNow is this run's reader, what the node-engine's verbs seat-list and seat-list-again answer from. +var seatListNow atomic.Pointer[seatListState] + +// seatListRead answers the verb seat-list: what the engine holds of the list and what it waits for (ADR 0636 §7). +func seatListRead() (any, error) { + st := seatListNow.Load() + if st == nil { + return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet") + } + status := st.reader.Status() + return map[string]any{"status": status, "health": status.Health()}, nil +} + +// seatListAgain answers the verb seat-list-again: the list read now from the seat, and what is held afterwards. +func seatListAgain() (any, error) { + st := seatListNow.Load() + if st == nil { + return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet") + } + err := st.reader.Reload() + st.waiter.Changed() // a read that changed nothing held still ends the wait + out, _ := seatListRead() + if err != nil { + out.(map[string]any)["error"] = err.Error() + } + return out, nil +} + +// seatListState is one run's reader, the finding that it waits, and the channel on which the composer hears a change. +type seatListState struct { + reader *seatlist.Reader + waiter *seatwait.Waiter + changed chan struct{} +} + +// sessionBus is the reader's bus over the link: a verb of the seat is asked in the engine's own name, an event heard on +// the seat's event subject for as long as ctx lives. +type sessionBus struct { + ctx context.Context + session *link.Session +} + +func (b sessionBus) Ask(key string, body any) (json.RawMessage, error) { + seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + return b.session.AskMeshSeat(b.ctx, seat, verb, body) +} + +func (b sessionBus) Subscribe(pattern string, handle func(string, json.RawMessage) error) error { + seat, event, _ := strings.Cut(pattern, ".") + heard, stop, err := b.session.Hear(seatdef.EventSubject(seat, event)) + if err != nil { + return err + } + go func() { + defer stop() + for { + select { + case <-b.ctx.Done(): + return + case body, ok := <-heard: + if !ok { + return + } + _ = handle(pattern, body) + } + } + }() + return nil +} + +// seatWaitClock is the clock of the node's bound; the tests give their own. +var seatWaitClock seatwait.Clock = realClock{} + +type realClock struct{} + +func (realClock) Now() time.Time { return time.Now() } +func (realClock) AfterFunc(d time.Duration, f func()) seatwait.Timer { + return time.AfterFunc(d, f) +} + +// findingsOfTheNode writes the node's findings in its state `findings`, in the layout of the seat mesh-conditions. +type findingsOfTheNode struct{ session *link.Session } + +func (w findingsOfTheNode) encode(f views.Finding) (string, []byte, error) { + key := conditions.Key(conditions.ScopeMachine, f.ID, f.Kind) + v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning, + Subject: condseat.Subject{Scope: conditions.ScopeMachine, ID: f.ID, Machine: f.ID}, Summary: f.Summary, + Headline: f.Headline, Explanation: f.Explanation, Source: "seat-list-unread", At: f.At.UTC().Format(time.RFC3339)} + if err := conditions.CheckFinding(v); err != nil { + return "", nil, err + } + raw, err := json.Marshal(v) + return key, raw, err +} + +func (w findingsOfTheNode) Put(f views.Finding) error { + key, raw, err := w.encode(f) + if err != nil { + return err + } + return w.session.PutFinding(key, raw) +} + +func (w findingsOfTheNode) Delete(f views.Finding) error { + key, _, err := w.encode(f) + if err != nil { + return err + } + return w.session.DeleteFinding(key) +} + +// startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what +// it holds again, saying so to the composer and to the waiter that writes the five-minute finding. +func startSeatList(ctx context.Context, session *link.Session, node string, say func(string)) (*seatListState, error) { + st := &seatListState{changed: make(chan struct{}, 1)} + var announce func() + reader, err := seatlist.Start(seatlist.Config{Bus: sessionBus{ctx, session}, Bootstrap: seatlist.Bootstrapped(), + OnChange: func(seats []catalogue.Seat) { + catalogue.UseSeats(seats) + if announce != nil { // the first read happens inside Start, before the waiter exists + announce() + } + }}) + if err != nil { + return nil, fmt.Errorf("the node-engine cannot follow the seat list: %w", err) + } + st.reader = reader + st.waiter = &seatwait.Waiter{Node: node, Observer: "node-engine", Reader: reader, Findings: findingsOfTheNode{session}, + Clock: seatWaitClock, Say: say} + announce = func() { + st.waiter.Changed() + select { + case st.changed <- struct{}{}: + default: + } + } + catalogue.UseSeats(reader.Seats()) + announce() + seatListNow.Store(st) + return st, nil +} diff --git a/go.mod b/go.mod index 1203a31..17392cf 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,8 @@ require ( require git.novox.be/novox/mesh-contracts/go v0.0.0-20261104174344-ae5266e77639 // indirect require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-20261105000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-20261105000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-20261104174344-ae5266e77639 git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 git.novox.be/novox/mesh-contracts/go/seats/meshvault v0.0.0-20261104174344-ae5266e77639 diff --git a/internal/link/findings.go b/internal/link/findings.go new file mode 100644 index 0000000..0b0ba3f --- /dev/null +++ b/internal/link/findings.go @@ -0,0 +1,63 @@ +package link + +import ( + "errors" + "fmt" + + "github.com/nats-io/nats.go" +) + +// FindingsBucket is the node-engine's state `findings`: the findings this engine writes in its own bucket, one key each, +// which the condition keeper raises as conditions and clears when the key goes (novox/hq ADR 0564 §2, ADR 0636 §1). +const FindingsBucket = "mesh-host_findings" + +func (l *natsLink) findingsKV() (nats.KeyValue, error) { + kv, err := l.js.KeyValue(FindingsBucket) + if err != nil { + return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's manifest; "+ + "written again once it is): %w", FindingsBucket, err) + } + return kv, nil +} + +func (q *Session) findings() (nats.KeyValue, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + l, ok := bus.(interface{ findingsKV() (nats.KeyValue, error) }) + if bus == nil || !ok { + return nil, ErrNoLink + } + return l.findingsKV() +} + +// PutFinding writes one finding of this engine over the link open now. +func (q *Session) PutFinding(key string, value []byte) error { + kv, err := q.findings() + if err != nil { + return err + } + _, err = kv.Put(key, value) + return err +} + +// DeleteFinding clears one finding of this engine over the link open now; a finding already gone is cleared. +func (q *Session) DeleteFinding(key string) error { + kv, err := q.findings() + if err != nil { + return err + } + if err = kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return err + } + return nil +} + +func (o OverNATS) findingsKV() (nats.KeyValue, error) { + kv, err := o.JS.KeyValue(FindingsBucket) + if err != nil { + return nil, fmt.Errorf("the node-engine's state %s cannot be bound: %w", FindingsBucket, err) + } + return kv, nil +} diff --git a/internal/link/handacts.go b/internal/link/handacts.go index a192515..bb0cf84 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -51,6 +51,11 @@ const ( ActComposeCandidate = nodeengine.VerbComposeCandidate // ActManifestReads reads the reader capabilities this engine states (novox/hq ADR 0618 §3). ActManifestReads = nodeengine.VerbManifestReads + // ActSeatList reads the seat list this engine holds and what it waits for; ActSeatListAgain makes it read the list now + // (novox/hq ADR 0636 §7). The node-engine seat's definition carries both verbs, which answer only the operator and the + // healers. + ActSeatList = "seat-list" + ActSeatListAgain = "seat-list-again" ) // The states of an act: asked, done and refused are also the seat's events. @@ -149,6 +154,9 @@ type HandActs struct { // Packages answers the node's key of the engine's state packages beside the controller's mirror's key and their // difference, and with write writes the key again from what the engine composed (the verb packages, ADR 0565 §6, §15). Packages func(write bool) (any, error) + // SeatList answers the seat list the engine holds; SeatListAgain makes it read the list now and answers what it holds. + SeatList func() (any, error) + SeatListAgain func() (any, error) // ComposeCandidate composes a check's candidate (the indexer's candidate key) on this node in shadow and answers per // module, applying and writing nothing; with input, also the composition input it composed from (the verb // compose-candidate, ADR 0593 §7). @@ -219,6 +227,16 @@ func (h *HandActs) Call(c EngineCall) (any, error) { return nil, errors.New("this engine composes nothing on a call yet: its node agent has not started") } return h.Compose(arg("module")) + case ActSeatList: + if h.SeatList == nil { + return nil, errors.New("this engine reads no seat list yet: its node agent has not started") + } + return h.SeatList() + case ActSeatListAgain: + if h.SeatListAgain == nil { + return nil, errors.New("this engine reads no seat list yet: its node agent has not started") + } + return h.SeatListAgain() case ActPackages: if h.Packages == nil { return nil, errors.New("this engine keeps no packages yet: its node agent has not started") diff --git a/internal/link/movesecret.go b/internal/link/movesecret.go index cb17df0..8819d8a 100644 --- a/internal/link/movesecret.go +++ b/internal/link/movesecret.go @@ -181,7 +181,12 @@ func CalledVaultSubject(verb, node string) string { // CalledVaultSubjectAs is CalledVaultSubject for the login the engine is dialled as (node. or node.--b, ADR 0583 §1): // the bus lets only that user publish a subject naming it. func CalledVaultSubjectAs(verb, user string) string { - return "mesh.seat." + VaultSeat + ".call." + verb + "." + strings.ReplaceAll(user, ".", "~") + return CalledSeatSubjectAs(VaultSeat, verb, user) +} + +// CalledSeatSubjectAs is where a verb of a seat held once for the mesh is asked naming the login the engine is dialled as. +func CalledSeatSubjectAs(seat, verb, user string) string { + return "mesh.seat." + seat + ".call." + verb + "." + strings.ReplaceAll(user, ".", "~") } // VaultWithin is how long the engine gives the vault to answer one step. @@ -190,6 +195,17 @@ const VaultWithin = 15 * time.Second // AskVault asks one of the vault's verbs in this node's engine's own name over the link open now, and answers its // result, or its refusal as an error. func (q *Session) AskVault(ctx context.Context, verb string, args any) (json.RawMessage, error) { + return q.askCalled(ctx, VaultSeat, verb, args, VaultWithin, "the vault's "+verb) +} + +// AskMeshSeat asks a verb of a seat held once for the mesh in this node's engine's own name over the link open now, and +// answers its result, or its refusal as an error: the way the engine reads the seat list from the seat mesh-catalog +// (novox/hq ADR 0636 §4). +func (q *Session) AskMeshSeat(ctx context.Context, seat, verb string, args any) (json.RawMessage, error) { + return q.askCalled(ctx, seat, verb, args, VaultWithin, seat+"."+verb) +} + +func (q *Session) askCalled(ctx context.Context, seat, verb string, args any, within time.Duration, what string) (json.RawMessage, error) { q.init() q.mu.Lock() bus := q.bus @@ -205,18 +221,18 @@ func (q *Session) AskVault(ctx context.Context, verb string, args any) (json.Raw if err != nil { return nil, err } - asking, cancel := context.WithTimeout(ctx, VaultWithin) + asking, cancel := context.WithTimeout(ctx, within) defer cancel() - reply, err := asker.Ask(asking, CalledVaultSubjectAs(verb, q.Membership.User()), body) + reply, err := asker.Ask(asking, CalledSeatSubjectAs(seat, verb, q.Membership.User()), body) if err != nil { - return nil, fmt.Errorf("the vault's %s could not be asked: %w", verb, err) + return nil, fmt.Errorf("%s could not be asked: %w", what, err) } var env struct { Result json.RawMessage `json:"result"` Error string `json:"error"` } if err := json.Unmarshal(reply, &env); err != nil { - return nil, fmt.Errorf("the vault answered %s with something that is not a tool's answer", verb) + return nil, fmt.Errorf("%s answered with something that is not a tool's answer", what) } if env.Error != "" { return nil, errors.New(env.Error) diff --git a/internal/nodegrant/testdata/node.json b/internal/nodegrant/testdata/node.json index 16ce048..bfb8c60 100644 --- a/internal/nodegrant/testdata/node.json +++ b/internal/nodegrant/testdata/node.json @@ -67,6 +67,7 @@ "$JS.API.STREAM.INFO.KV_mesh-delivery-indexer_verdicts", "$JS.API.STREAM.INFO.KV_mesh-host_acts", "$JS.API.STREAM.INFO.KV_mesh-host_data", + "$JS.API.STREAM.INFO.KV_mesh-host_findings", "$JS.API.STREAM.INFO.KV_mesh-host_health", "$JS.API.STREAM.INFO.KV_mesh-host_packages", "$JS.API.STREAM.INFO.KV_mesh-provisioner_provisions", @@ -91,6 +92,7 @@ "$KV.composed-NODE.>", "$KV.mesh-host_acts.NODE.>", "$KV.mesh-host_data.NODE.>", + "$KV.mesh-host_findings.>", "$KV.mesh-host_health.NODE", "$KV.mesh-host_packages.NODE", "$KV.node-engine_have.NODE.>", @@ -100,6 +102,7 @@ "$SRV.PING.>", "$SRV.PING.node-tools.NODE", "mesh.control.NODE.>", + "mesh.seat.mesh-catalog.call.seat-definitions.node~NODE", "mesh.seat.node-backup.call.backed-up.NODE.node~NODE", "mesh.seat.node-backup.call.now.NODE.node~NODE", "mesh.seat.node-backup.tool.backed-up.NODE", @@ -130,8 +133,10 @@ "mesh.node.NODE.ask.report", "mesh.node.NODE.ask.setuid-search", "mesh.node.NODE.declare", + "mesh.seat.mesh-catalog.event.seat-definition-changed", + "mesh.seat.mesh-catalog.event.seat-definition-removed", "mesh.seat.node-engine.call.*.NODE.*", "mesh.seat.node-engine.tool.*.NODE" ], "AllowResponses": true -} \ No newline at end of file +} From f29b19aff229fc1c5545c95da7d745d22e445719 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 14:30:08 +0100 Subject: [PATCH 5/5] Node-engine manifest: invokes the seat mesh-catalog's seat-definitions, consumes its two events, keeps the state findings (H-7, ADR 0636; issue 1966) --- module.json | 8 ++++++++ packaging/operator_keys_test.go | 15 ++++++++++----- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/module.json b/module.json index b71573b..2f3da06 100644 --- a/module.json +++ b/module.json @@ -14,13 +14,21 @@ } ] }, + "invokes": [ + "seat:mesh-catalog.seat-definitions" + ], "consumes": [ + "mesh-catalog.seat-definition-changed", + "mesh-catalog.seat-definition-removed", "restic.backed-up" ], "state": [ { "name": "health" }, + { + "name": "findings" + }, { "name": "data" }, diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 5c77c08..487579a 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -3,6 +3,7 @@ package packaging_test import ( "encoding/json" "os" + "slices" "strings" "testing" @@ -65,8 +66,9 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - if len(m.Consumes) != 1 || m.Consumes[0] != "restic.backed-up" { - t.Fatalf("consumes %v", m.Consumes) + // The backup event, and the two events of the seat that serves the seat list (ADR 0636 §4). + if want := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "restic.backed-up"}; !slices.Equal(m.Consumes, want) { + t.Fatalf("consumes %v, want %v", m.Consumes, want) } // 774: the acts bucket mesh-host_acts is not declared as state for now: a declared state of a module with no account // of its own is carried by every node's tool runner, which would be a second writer of the engine's own keys and @@ -87,9 +89,12 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { // ADR 0577 clause 1: and the state data, one key per module placed on the node that declares data items, kept until // the module leaves the node; created by the grants part from this manifest as health is. // ADR 0565 clause 6: and the state packages, one key per node, its compositions' declared packages, read by pacman. - if len(parsed.State) != 3 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 || - parsed.State[1].Name != "data" || parsed.State[1].TTLSeconds != 0 || parsed.State[2].Name != "packages" || - parsed.State[2].TTLSeconds != 0 { + // ADR 0564 §2, ADR 0636 §1: and the state findings, the node's finding that it waits for the seat list (no TTL: the + // finding stays until the first full read deletes its key). + if len(parsed.State) != 4 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 || + parsed.State[1].Name != "findings" || parsed.State[1].TTLSeconds != 0 || + parsed.State[2].Name != "data" || parsed.State[2].TTLSeconds != 0 || parsed.State[3].Name != "packages" || + parsed.State[3].TTLSeconds != 0 { t.Fatalf("the manifest declares the state %+v", parsed.State) } }