diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index ea3f3b4a..2b5b8b9b 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -20,7 +20,6 @@ import ( "git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay" "git.novox.be/novox/mesh-sdk/go/declaration" - "git.novox.be/novox/mesh-sdk/go/views" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/compose/frombus" @@ -593,7 +592,17 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus 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). - findings := newFindingsOfTheNode(session, node) + // The finding writer is started first, so it holds what an earlier run left before the reader writes or clears any. + findings, ferr := newFindingsOfTheNode(session, node, say) + if ferr != nil { // a writer that cannot be made is a fault of its configuration: loud, never a node without findings + say("the node-engine cannot make its finding writer, so its node agent stops: " + ferr.Error()) + stopRun() + cancel() + return + } + if werr := findings.Start(run); werr != nil { + say("the node-engine's findings are kept in its state and not said as events until it is started again: " + werr.Error()) + } seats, err := startSeatList(run, session, node, findings, say) if err != nil { // No reader is no gate that lets everything through: the node acts on nothing new, as one that waits does. @@ -601,11 +610,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus } // The marks of the haves run off the composition pass, each bounded (see marker). marks := newMarker(run) - if werr := findings.Start(run, say); werr != nil { - say("the node-engine's findings are kept in its state and not said as events until it is started again: " + werr.Error()) - } - heldFindings := &seatwait.HeldKeeper{Node: node, Observer: findingsModule, Writer: findings, - Left: func() ([]views.Finding, error) { return findings.left(node) }, Say: say} + heldFindings := &seatwait.HeldKeeper{Node: node, Observer: findingsModule, Writer: findings, Ctx: run, Say: say} hooks := composerHooks{ seatList: seats, seatListFailed: err != nil, diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 226c0e1b..fc86409e 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -992,6 +992,11 @@ func (c *fiveMinutes) pass() { } } +// findingSubject is the subject of one of the engine's finding events on node (views.FindingSubjects). +func findingSubject(event, node string) string { + return "mesh.stream." + findingsModule + ".event." + event + "." + node +} + // 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 diff --git a/cmd/mesh-host/seatlist.go b/cmd/mesh-host/seatlist.go index c53d423c..c06f5a9f 100644 --- a/cmd/mesh-host/seatlist.go +++ b/cmd/mesh-host/seatlist.go @@ -5,14 +5,10 @@ import ( "encoding/json" "errors" "fmt" - "slices" - "sort" "strings" - "sync" "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" @@ -110,179 +106,49 @@ func (realClock) AfterFunc(d time.Duration, f func()) seatwait.Timer { // their observer (module.json's `module` and its stream `mesh-host-findings`, design 47 G §383). const findingsModule = "mesh-host" -// findingsOfTheNode keeps the node's findings through mesh-sdk's finding writer (design 47 G §383, §384, §461): each finding is -// written in the node's state `findings` and said as an event on the engine's own stream, the engine says every finding it -// observes at its start and when the keeper asks, and the keeper reads no state of the engine's. +// nodeFindings is the node-engine's finding writer: mesh-sdk's held writer, for the engine's findings are raised as they +// happen and not judged afresh (design 47 G A-29, A-40), with the contract of the seat mesh-conditions. +type nodeFindings = views.HeldWriter[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent] + +// findingsOfTheNode keeps the node's findings through mesh-sdk's held writer (design 47 G §383, §384, §461, A-29): each +// finding is written in the node's state `findings` and said as an event on the engine's own stream (a foundation part's +// `mesh-host-findings`, §529, A-30), the engine says what it holds at its start and when the keeper asks, and the keeper +// reads no state of the engine's. type findingsOfTheNode struct { - session *link.Session - node string - - mu sync.Mutex - ctx context.Context - writer *views.Writer + *nodeFindings } -func newFindingsOfTheNode(session *link.Session, node string) *findingsOfTheNode { - return &findingsOfTheNode{session: session, node: node} -} - -// findingSubject is a stream event's subject for the observer's own node (design 47 G §383, ADR 0363 §2). -func findingSubject(event, node string) string { - return "mesh.stream." + findingsModule + ".event." + event + "." + node -} - -// Start makes the writer and states the findings now, until ctx ends. A writer that cannot start (the link has no stream -// yet) is said; the findings are then kept in the state alone, and said by the next start. -func (w *findingsOfTheNode) Start(ctx context.Context, say func(string)) error { - writer, err := views.NewWriter(views.WriterConfig{ - Module: findingsModule, - State: link.FindingsState{Session: w.session}, - Subjects: views.WriterSubjects{Written: findingSubject("finding-written", w.node), Cleared: findingSubject("finding-cleared", w.node), - Stated: findingSubject("findings-stated", w.node), Asked: condseat.SubjectFindingsAsked}, - Observe: w.observe, +// newFindingsOfTheNode makes the writer at once; it reads the state when it is started (A-40). Its events are published on +// the engine's own stream awaiting the bus's acknowledgement and its consumer of `findings-asked` is its own. +func newFindingsOfTheNode(session *link.Session, node string, say func(string)) (*findingsOfTheNode, error) { + writer, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ + Module: findingsModule, + Node: node, + State: link.FindingsState{Session: session}, + Kinds: []string{seatwait.Kind, seatwait.HeldKind}, + Asked: condseat.SubjectFindingsAsked, Logf: func(format string, args ...any) { say(fmt.Sprintf(format, args...)) }, - Publish: w.session.PublishEvent, + Publish: session.PublishEvent, Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) { - return w.session.ListenFrom(condseat.StreamFindingsAsked, subject, handle) + return session.ListenFrom(condseat.StreamFindingsAsked, subject, handle) }, }) - if err != nil { - return err - } - w.mu.Lock() - w.ctx, w.writer = ctx, writer - w.mu.Unlock() - go func() { <-ctx.Done(); writer.Close() }() - return writer.Start(ctx) -} - -func (w *findingsOfTheNode) started() (context.Context, *views.Writer, error) { - w.mu.Lock() - defer w.mu.Unlock() - if w.writer == nil { - return nil, nil, errors.New("the node-engine's finding writer is not started") - } - return w.ctx, w.writer, nil -} - -// findingBody is the finding in the layout of the seat mesh-conditions (design 47 G §439). The seat list's finding is of the -// machine; a held module's is of the module on a machine, `.`. The writer names the observer. -func findingBody(f views.Finding) views.FindingBody { - scope, machine := conditions.ScopeMachine, f.ID - if f.Kind == seatwait.HeldKind { - scope = conditions.ScopeModule - machine = f.ID[strings.LastIndex(f.ID, ".")+1:] - } - return views.FindingBody{Kind: f.Kind, Scope: scope, ID: f.ID, At: f.At.UTC(), Severity: conditions.Warning, - Summary: f.Summary, Headline: f.Headline, Explanation: f.Explanation, Source: f.Kind, - Subject: views.FindingSubject{Scope: scope, ID: f.ID, Machine: machine}} -} - -// left is the findings of held modules an earlier run of this engine left on node: its keys of this kind. -func (w *findingsOfTheNode) left(node string) ([]views.Finding, error) { - keys, err := w.session.FindingKeys() if err != nil { return nil, err } - var out []views.Finding - for _, k := range keys { - id, ok := strings.CutPrefix(k, conditions.ScopeModule+".") - if !ok { - continue - } - if id, ok = strings.CutSuffix(id, "."+seatwait.HeldKind); !ok || !strings.HasSuffix(id, "."+node) { - continue - } - out = append(out, views.Finding{Observer: findingsModule, Kind: seatwait.HeldKind, ID: id}) - } - return out, nil + return &findingsOfTheNode{writer}, nil } -func (w *findingsOfTheNode) Put(f views.Finding) error { - ctx, writer, err := w.started() - if err != nil { - return err - } - return writer.Write(ctx, findingBody(f)) -} - -// keyOf is the finding's key in the state; a read or a clear needs no more of the finding than Kind and ID. -func (w *findingsOfTheNode) keyOf(f views.Finding) string { - b := findingBody(f) - return b.Key() -} - -func (w *findingsOfTheNode) Delete(f views.Finding) error { - ctx, writer, err := w.started() - if err != nil { - // Before the writer starts, which is at its start the one statement that says what this engine observes: the key goes, - // and that statement names no finding of it, so the keeper clears it (§384). - return w.session.DeleteFinding(w.keyOf(f)) - } - b := findingBody(f) - why := "the node-engine no longer observes it" - switch f.Kind { - case seatwait.Kind: - why = "the node-engine has read the whole seat list" - case seatwait.HeldKind: - why = "the module is composed again or is no longer wanted on the node" - } - return writer.Clear(ctx, b.Scope, b.ID, b.Kind, why) -} - -// Has says whether the finding stands in the node's state `findings` now. -func (w *findingsOfTheNode) Has(f views.Finding) (bool, error) { - return w.session.HasFinding(w.keyOf(f)) -} - -// observe is the node-engine's own truth for a statement (§384): the findings that stand in its state and are still true -// (the seat list's only while the reader waits for it, so an engine started with the full list states none), and the seat -// list's finding as pending while its five minutes run and none stands yet. -func (w *findingsOfTheNode) observe(context.Context) (views.Observed, error) { - var out views.Observed - st := seatListNow.Load() - waits := st != nil && seatwait.Waits(st.reader.Status()) - keys, err := w.session.FindingKeys() - if err != nil { - return out, err - } - sort.Strings(keys) - for _, key := range keys { - scope, rest, ok := strings.Cut(key, ".") - i := strings.LastIndex(rest, ".") - if !ok || i < 0 { - continue - } - id, kind := rest[:i], rest[i+1:] - if kind == seatwait.Kind && !waits { - continue // read in full: it is no finding of this engine now, and its key goes with this statement - } - e, err := link.FindingsState{Session: w.session}.Get(key) - if err != nil || e == nil { - continue - } - var body views.FindingBody - if json.Unmarshal(e.Value, &body) != nil { - continue - } - body.Scope, body.ID, body.Kind = scope, id, kind - out.Findings = append(out.Findings, body) - } - if waits { - if stands, since := st.waiter.Waiting(); stands { - key := conditions.Key(conditions.ScopeMachine, w.node, seatwait.Kind) - if !slices.Contains(keys, key) { - out.Pending = append(out.Pending, views.PendingKey{Key: key, Since: since.UTC(), Bound: seatwait.Bound.String()}) - } - } - } - return out, nil +// Start seeds what the state holds, states the findings now and hears the keeper until ctx ends. +func (w *findingsOfTheNode) Start(ctx context.Context) error { + go func() { <-ctx.Done(); w.Close() }() + return w.nodeFindings.Start(ctx) } // 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, and stating again what // the engine reads (link.Session.RestateReads, ADR 0618 §4): the seats it states are those of the list it holds. -func startSeatList(ctx context.Context, session *link.Session, node string, findings seatwait.Findings, say func(string)) (*seatListState, error) { +func startSeatList(ctx context.Context, session *link.Session, node string, findings seatwait.Writer, say func(string)) (*seatListState, error) { st := &seatListState{changed: make(chan struct{}, 1)} // Start returns at once and reads the seat on a goroutine, which may announce a change before the waiter exists. var announce atomic.Pointer[func()] @@ -300,7 +166,7 @@ func startSeatList(ctx context.Context, session *link.Session, node string, find return nil, fmt.Errorf("the node-engine cannot follow the seat list: %w", err) } st.reader = reader - st.waiter = &seatwait.Waiter{Node: node, Observer: findingsModule, Reader: reader, Findings: findings, + st.waiter = &seatwait.Waiter{Node: node, Observer: findingsModule, Reader: reader, Writer: findings, Ctx: ctx, Clock: seatWaitClock, Say: say} told := func() { st.waiter.Changed() diff --git a/cmd/mesh-host/seatlist_unread_test.go b/cmd/mesh-host/seatlist_unread_test.go index a9a8cb18..ef6c1208 100644 --- a/cmd/mesh-host/seatlist_unread_test.go +++ b/cmd/mesh-host/seatlist_unread_test.go @@ -3,8 +3,6 @@ package main import ( "testing" - "git.novox.be/novox/mesh-sdk/go/views" - "github.com/novox/mesh-host/internal/seatwait" "github.com/novox/mesh-host/internal/selfcompose" ) @@ -21,10 +19,4 @@ func TestAReaderThatCouldNotStartHoldsTheComposer(t *testing.T) { if got := composerEnv(nil, "n1", func(string) {}, composerHooks{}).SeatList; got != nil { t.Errorf("a composer with no reader asked for gates by %v, want none (the shadow)", got) } - // The finding of a held module is of the module on its machine, in a layout the condition keeper accepts (design 47, test 55). - body := findingBody(views.Finding{Observer: "node-engine", Kind: seatwait.HeldKind, - ID: seatwait.HeldID("mesh-grants", "ace"), Summary: "mesh-grants is held on ace: x", Headline: "h", Explanation: "e"}) - if body.Key() != "module.mesh-grants.ace.module-held" || body.Subject.Machine != "ace" || body.Observer != "" { - t.Errorf("a held module's finding is %+v with key %q", body, body.Key()) - } } diff --git a/internal/link/findings.go b/internal/link/findings.go index ab636c1a..0a9a0d9f 100644 --- a/internal/link/findings.go +++ b/internal/link/findings.go @@ -46,21 +46,6 @@ func (q *Session) PutFinding(key string, value []byte) error { return err } -// HasFinding says whether one finding of this engine stands in its state now. -func (q *Session) HasFinding(key string) (bool, error) { - kv, err := q.findings() - if err != nil { - return false, err - } - if _, err = kv.Get(key); err != nil { - if errors.Is(err, nats.ErrKeyNotFound) { - return false, nil - } - return false, err - } - return true, nil -} - // FindingKeys lists the keys of this engine's findings over the link open now; none when the bucket holds none. func (q *Session) FindingKeys() ([]string, error) { kv, err := q.findings() diff --git a/internal/seatwait/held.go b/internal/seatwait/held.go index b2682d50..5ac18e8b 100644 --- a/internal/seatwait/held.go +++ b/internal/seatwait/held.go @@ -6,34 +6,32 @@ package seatwait import ( + "context" "fmt" "sort" "strings" - "sync" "time" - "git.novox.be/novox/mesh-sdk/go/views" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" ) // HeldKind is the finding of a held module; its id is `.`. const HeldKind = "module-held" -// HeldKeeper writes one finding for each module the node holds, and clears it when the module is no longer held. +// HeldKeeper writes one finding for each module the node holds, and clears it when the module is no longer held. What it +// has written is what its writer holds (design 47 G A-29, A-38, A-40): the findings an earlier run of the engine left are +// seeded into it at its Start, so a module held then and composed while the engine was away loses its condition at the +// first Reconcile, and the keeper keeps no set of its own. type HeldKeeper struct { Node string Observer string - Writer views.FindingWriter - // Left lists the findings of this kind an earlier run of the engine left, once, at the first Reconcile: a module held - // then and composed while the engine was away must not keep its condition. Nil leaves none. - Left func() ([]views.Finding, error) + Writer Writer + // Ctx is the run's context for the writer's acts; nil is the background. + Ctx context.Context // Now is the clock; nil is time.Now. Now func() time.Time // Say tells the node's journal; nil says nothing. Say func(string) - - mu sync.Mutex - swept bool - written map[string]views.Finding // by module } // HeldID is the finding's id for a module on a node. @@ -43,29 +41,21 @@ func HeldID(module, node string) string { return module + "." + node } // held for other words, and one is cleared for a module no longer held, composed or no longer wanted here. A write or a // clear that fails is said and tried again at the next Reconcile. func (k *HeldKeeper) Reconcile(held map[string]string) { - k.mu.Lock() - defer k.mu.Unlock() - if k.written == nil { - k.written = map[string]views.Finding{} + ctx := k.Ctx + if ctx == nil { + ctx = context.Background() } - if !k.swept && k.Left != nil { - left, err := k.Left() - if err != nil { - k.say(fmt.Sprintf("the findings an earlier run left of held modules cannot be read: %v", err)) - } else { - k.swept = true - for _, f := range left { - module := strings.TrimSuffix(f.ID, "."+k.Node) - if _, ok := k.written[module]; !ok { - k.written[module] = f - } - } + written := map[string]condseat.Finding{} // by module, what the writer holds of this node's held modules + for _, f := range k.Writer.Held() { + if f.Kind != HeldKind || f.Scope != scopeModule { + continue + } + if module, ok := strings.CutSuffix(f.ID, "."+k.Node); ok { + written[module] = f } - } else if k.Left == nil { - k.swept = true } - names := make([]string, 0, len(k.written)) - for m := range k.written { + names := make([]string, 0, len(written)) + for m := range written { names = append(names, m) } sort.Strings(names) @@ -73,11 +63,9 @@ func (k *HeldKeeper) Reconcile(held map[string]string) { if _, still := held[m]; still { continue } - if err := k.Writer.Delete(k.written[m]); err != nil { + if err := k.Writer.Clear(ctx, scopeModule, written[m].ID, HeldKind, "the module is composed again or is no longer wanted on the node"); err != nil { k.say(fmt.Sprintf("the finding that %s is held on %s cannot be cleared: %v", m, k.Node, err)) - continue } - delete(k.written, m) } names = names[:0] for m := range held { @@ -86,19 +74,21 @@ func (k *HeldKeeper) Reconcile(held map[string]string) { sort.Strings(names) for _, m := range names { why := held[m] - if was, ok := k.written[m]; ok && was.Summary != "" && strings.HasSuffix(was.Summary, ": "+why) { + if was, ok := written[m]; ok && strings.HasSuffix(was.Summary, ": "+why) { continue } - f := views.Finding{Observer: k.Observer, Kind: HeldKind, ID: HeldID(m, k.Node), + id := HeldID(m, k.Node) + f := condseat.Finding{Observer: k.Observer, Kind: HeldKind, Scope: scopeModule, ID: id, Severity: condseat.FindingSeverityWarning, + Source: HeldKind, + Subject: condseat.Subject{Scope: scopeModule, ID: id, Machine: k.Node}, Summary: fmt.Sprintf("%s is held on %s: %s", m, k.Node, why), Headline: fmt.Sprintf("%s is held on %s", m, k.Node), Explanation: "The node leaves this module out of its pass, keeps what runs of it and composes and applies every other module. The finding clears when the module is composed again or is no longer wanted on the node.", At: k.now()} - if err := k.Writer.Put(f); err != nil { + if err := k.Writer.Write(ctx, f); err != nil { k.say(fmt.Sprintf("the finding that %s is held on %s cannot be written: %v", m, k.Node, err)) continue } - k.written[m] = f k.say(f.Summary) } } diff --git a/internal/seatwait/seatwait.go b/internal/seatwait/seatwait.go index 54a8f4f9..7fc83a44 100644 --- a/internal/seatwait/seatwait.go +++ b/internal/seatwait/seatwait.go @@ -8,12 +8,13 @@ package seatwait import ( + "context" "fmt" "sync" "time" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "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). @@ -36,12 +37,28 @@ type Clock interface { AfterFunc(d time.Duration, f func()) Timer } -// Findings is the node's own state `findings`: the writer, and the read of whether a finding stands there. The bus state -// is the one truth of whether the finding exists (ADR 0636 §1, issue 1966): the Waiter keeps no copy, so a new process -// of the node-engine clears what an earlier one wrote. -type Findings interface { - views.FindingWriter - Has(f views.Finding) (bool, error) +// Writer is the node-engine's finding writer as the keepers use it: mesh-sdk's held writer over the node's state `findings` +// (views.HeldWriter, design 47 G A-29, A-38, A-40), which holds what stands there from its start, so the Waiter keeps no +// copy and a new process of the node-engine clears what an earlier one wrote (ADR 0636 §1, issue 1966). +type Writer interface { + Write(ctx context.Context, f condseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []condseat.Finding +} + +const ( + scopeMachine = "machine" + scopeModule = "module" +) + +// stands is whether the writer holds the finding of that scope, id and kind. +func stands(w Writer, scope, id, kind string) bool { + for _, f := range w.Held() { + if f.Scope == scope && f.ID == id && f.Kind == kind { + return true + } + } + return false } // Waiter writes and clears the node's finding. @@ -49,8 +66,10 @@ type Waiter struct { Node string Observer string Reader Reader - Findings Findings + Writer Writer Clock Clock + // Ctx is the run's context for the writer's acts; nil is the background. + Ctx context.Context // Say tells the node's journal; nil says nothing. Say func(string) @@ -59,11 +78,15 @@ type Waiter struct { since time.Time } -// key is the finding this node's Waiter writes: Kind and ID name it, the rest is its text. -func (w *Waiter) key() views.Finding { - return views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node} +func (w *Waiter) ctx() context.Context { + if w.Ctx != nil { + return w.Ctx + } + return context.Background() } +func (w *Waiter) stands() bool { return stands(w.Writer, scopeMachine, w.Node, Kind) } + func waiting(st seatlist.Status) bool { return !st.Heard || len(st.Waiting) > 0 } // Waits is whether a reader of that status holds only the definitions it started with, or waits for some. @@ -81,13 +104,8 @@ func (w *Waiter) Changed() { w.timer = nil } w.since = time.Time{} - stands, err := w.Findings.Has(w.key()) - if err != nil { - w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be read: %v", err)) - return - } - if stands { - if err := w.Findings.Delete(w.key()); err != nil { + if w.stands() { + if err := w.Writer.Clear(w.ctx(), scopeMachine, w.Node, Kind, "the node-engine has read the whole seat list"); err != nil { w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be cleared: %v", err)) } } @@ -96,7 +114,7 @@ func (w *Waiter) Changed() { if w.timer != nil { return } - if stands, err := w.Findings.Has(w.key()); err == nil && stands { + if w.stands() { return // an earlier process wrote it and the list is still unread: it stands until the first full read } w.since = w.Clock.Now() @@ -110,8 +128,7 @@ func (w *Waiter) Waiting() (bool, time.Time) { if w.timer != nil { return true, w.since } - stands, err := w.Findings.Has(w.key()) - return err == nil && stands, w.since + return w.stands(), w.since } func (w *Waiter) expired() { @@ -123,15 +140,16 @@ func (w *Waiter) expired() { w.since = time.Time{} return } - if stands, err := w.Findings.Has(w.key()); err == nil && stands { + if w.stands() { return } - f := views.Finding{Observer: w.Observer, Kind: Kind, ID: w.Node, + f := condseat.Finding{Observer: w.Observer, Kind: Kind, Scope: scopeMachine, ID: w.Node, Severity: condseat.FindingSeverityWarning, + Source: Kind, Subject: condseat.Subject{Scope: scopeMachine, ID: w.Node, Machine: 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 { + if err := w.Writer.Write(w.ctx(), f); err != nil { w.say(fmt.Sprintf("the node's finding that its seat list is unread cannot be written: %v", err)) return } diff --git a/internal/seatwait/seatwait_test.go b/internal/seatwait/seatwait_test.go index 3e945bd9..80f0a534 100644 --- a/internal/seatwait/seatwait_test.go +++ b/internal/seatwait/seatwait_test.go @@ -1,12 +1,13 @@ package seatwait import ( + "context" "errors" "testing" "time" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/seatlist" - "git.novox.be/novox/mesh-sdk/go/views" ) type fakeReader struct{ st seatlist.Status } @@ -45,25 +46,37 @@ func (c *fakeClock) advance(d time.Duration) { } } +// fakeFindings is a held writer: what it holds is what stands in the state, seeded by the test for an earlier process's. type fakeFindings struct { - put, deleted []views.Finding - err error - standing bool // the bus state holds the finding, written by this process or an earlier one + put []condseat.Finding + cleared []condseat.Finding // as held when cleared + why []string + err error + held []condseat.Finding } -func (f *fakeFindings) Has(views.Finding) (bool, error) { return f.standing, nil } +func (f *fakeFindings) Held() []condseat.Finding { return append([]condseat.Finding(nil), f.held...) } -func (f *fakeFindings) Put(x views.Finding) error { +func (f *fakeFindings) Write(_ context.Context, x condseat.Finding) error { if f.err != nil { return f.err } f.put = append(f.put, x) - f.standing = true + f.held = append(f.held, x) return nil } -func (f *fakeFindings) Delete(x views.Finding) error { - f.deleted = append(f.deleted, x) - f.standing = false + +func (f *fakeFindings) Clear(_ context.Context, scope, id, kind, why string) error { + for i, h := range f.held { + if h.Scope == scope && h.ID == id && h.Kind == kind { + f.held = append(f.held[:i:i], f.held[i+1:]...) + f.cleared = append(f.cleared, h) + f.why = append(f.why, why) + return nil + } + } + f.cleared = append(f.cleared, condseat.Finding{Scope: scope, ID: id, Kind: kind}) // cleared of nothing, said all the same + f.why = append(f.why, why) return nil } @@ -73,7 +86,7 @@ func caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead(t 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 := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Writer: findings, Clock: clock, Say: func(s string) { said = append(said, s) }} w.Changed() clock.advance(Bound - time.Second) @@ -92,8 +105,8 @@ func caseTheNodeWritesOneFindingAfterFiveMinutesAndClearsItAtTheFirstFullRead(t 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 len(findings.cleared) != 1 || findings.cleared[0].ID != "ace" { + t.Fatalf("the first full read left the finding: deleted %+v", findings.cleared) } if on, _ := w.Waiting(); on { t.Error("the waiter still waits after the full read") @@ -105,14 +118,14 @@ 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 := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Writer: 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) + if len(findings.put) != 0 || len(findings.cleared) != 0 { + t.Fatalf("a list read in time wrote %+v and cleared %+v", findings.put, findings.cleared) } // The seat answered the list already held: no change was said, so the timer is the one to look again. @@ -131,7 +144,7 @@ func caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain(t *testing.T) { 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 := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Writer: findings, Clock: clock, Say: func(s string) { said = append(said, s) }} w.Changed() clock.advance(Bound) if len(said) != 1 || len(findings.put) != 0 { @@ -150,8 +163,8 @@ func caseAFindingThatCannotBeWrittenIsSaidAndTriedAgain(t *testing.T) { func caseAFindingOfAnEarlierProcessIsClearedAtTheFirstFullRead(t *testing.T) { reader := &fakeReader{st: seatlist.Status{Heard: false}} clock := &fakeClock{now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)} - findings := &fakeFindings{standing: true} - w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Findings: findings, Clock: clock} + findings := &fakeFindings{held: []condseat.Finding{{Scope: "machine", ID: "ace", Kind: Kind}}} + w := &Waiter{Node: "ace", Observer: "node-engine", Reader: reader, Writer: findings, Clock: clock} w.Changed() clock.advance(Bound) if len(findings.put) != 0 || len(clock.timers) != 0 { @@ -162,34 +175,42 @@ func caseAFindingOfAnEarlierProcessIsClearedAtTheFirstFullRead(t *testing.T) { } reader.st = seatlist.Status{Heard: true} w.Changed() - if len(findings.deleted) != 1 || findings.deleted[0].ID != "ace" || len(findings.put) != 0 { - t.Fatalf("the first full read left the earlier process's finding: deleted %+v, put %+v", findings.deleted, findings.put) + if len(findings.cleared) != 1 || findings.cleared[0].ID != "ace" || len(findings.put) != 0 { + t.Fatalf("the first full read left the earlier process's finding: deleted %+v, put %+v", findings.cleared, findings.put) } } // Design 47, test 55: a module held for its claim is a finding that names the node, the module and why, written once, // and cleared when the module is composed again or no longer wanted, also one an earlier run of the engine left. func caseAHeldModuleIsAFindingClearedWhenItIsComposedOrNoLongerWanted(t *testing.T) { - findings := &fakeFindings{} - k := &HeldKeeper{Node: "ace", Observer: "node-engine", Writer: findings, - Left: func() ([]views.Finding, error) { - return []views.Finding{{Kind: HeldKind, ID: HeldID("gone", "ace")}}, nil - }} + // the writer holds what an earlier run of the engine left (seeded at its Start), and what another node's keeper wrote + findings := &fakeFindings{held: []condseat.Finding{ + {Scope: "module", ID: HeldID("gone", "ace"), Kind: HeldKind}, + {Scope: "module", ID: HeldID("elsewhere", "g14"), Kind: HeldKind}}} + k := &HeldKeeper{Node: "ace", Observer: "node-engine", Writer: findings} k.Reconcile(map[string]string{"mesh-grants": "its seat is refused", "other": "no claim"}) - if len(findings.put) != 2 || findings.put[0].ID != "mesh-grants.ace" || len(findings.deleted) != 1 || findings.deleted[0].ID != "gone.ace" { - t.Fatalf("wrote %v, cleared %v", findings.put, findings.deleted) + if len(findings.put) != 2 || findings.put[0].ID != "mesh-grants.ace" || len(findings.cleared) != 1 || findings.cleared[0].ID != "gone.ace" { + t.Fatalf("wrote %v, cleared %v", findings.put, findings.cleared) + } + // the finding is of the module on its machine, in the contract's layout (design 47, test 55) + if f := findings.put[0]; f.Scope != "module" || f.Subject.Scope != "module" || f.Subject.Machine != "ace" || f.Subject.ID != "mesh-grants.ace" || + f.Severity != condseat.FindingSeverityWarning || f.Source != HeldKind || f.Observer != "node-engine" { + t.Fatalf("a held module's finding is %+v", f) + } + if got := findings.held[0]; got.ID != "elsewhere.g14" { + t.Fatalf("another node's finding was touched: %+v", findings.held) } k.Reconcile(map[string]string{"mesh-grants": "its seat is refused", "other": "no claim"}) if len(findings.put) != 2 { t.Fatalf("a module held for the same words was written again: %v", findings.put) } k.Reconcile(map[string]string{"mesh-grants": "its seat is refused"}) - if len(findings.deleted) != 2 || findings.deleted[1].ID != "other.ace" { - t.Fatalf("a module composed again kept its finding: %v", findings.deleted) + if len(findings.cleared) != 2 || findings.cleared[1].ID != "other.ace" { + t.Fatalf("a module composed again kept its finding: %v", findings.cleared) } k.Reconcile(nil) - if len(findings.deleted) != 3 { - t.Fatalf("a module no longer wanted kept its finding: %v", findings.deleted) + if len(findings.cleared) != 3 { + t.Fatalf("a module no longer wanted kept its finding: %v", findings.cleared) } }