diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 5eecb947..ea3f3b4a 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -593,15 +593,18 @@ 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). - seats, err := startSeatList(run, session, node, say) + findings := newFindingsOfTheNode(session, node) + 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. say(err.Error() + "; the node composes nothing new until it is started again and reads the list (novox/hq ADR 0636 §1)") } // The marks of the haves run off the composition pass, each bounded (see marker). marks := newMarker(run) - findings := findingsOfTheNode{session} - heldFindings := &seatwait.HeldKeeper{Node: node, Observer: "node-engine", Writer: findings, + 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} hooks := composerHooks{ seatList: seats, diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 44c753ad..226c0e1b 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -11,6 +11,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" @@ -159,6 +160,13 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed { if b.findings, err = b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: link.FindingsBucket, History: 1}); err != nil { t.Fatal(err) } + // The engine's own finding stream and the keeper's ask stream (design 47 G §383, §390), made as the bus module makes them. + for _, c := range []nats.StreamConfig{{Name: "MESH-HOST-FINDINGS", Subjects: []string{"mesh.stream.mesh-host.event.>"}}, + {Name: condseat.StreamFindingsAsked, Subjects: []string{condseat.SubjectFindingsAsked}}} { + if _, err = b.js.AddStream(&c); 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 { @@ -1037,6 +1045,27 @@ func TestNatsANodeThatCannotReadTheSeatListActsOnNothingNew(t *testing.T) { if b.file("web") != "" { t.Fatalf("the node applied web with its finding written: %q", b.file("web")) } + // Design 47 G §383, §384, §390 (test 107): the key is followed by the event on the engine's own stream, the engine stated + // its findings at its start, and an ask of the keeper is answered by a new statement. + lastEvent := func(event string) (string, bool) { + m, err := b.js.GetLastMsg("MESH-HOST-FINDINGS", findingSubject(event, itNode)) + if err != nil { + return "", false + } + return string(m.Data), true + } + within(t, 5*time.Second, "the node did not say finding-written after its key", func() bool { + got, ok := lastEvent("finding-written") + return ok && strings.Contains(got, `"kind":"seat-list-unread"`) && strings.Contains(got, `"observer":"mesh-host"`) + }) + within(t, 5*time.Second, "the node did not state its findings at its start", func() bool { _, ok := lastEvent("findings-stated"); return ok }) + if _, err := b.js.Publish(condseat.SubjectFindingsAsked, []byte(`{"ask":"a1","at":"2026-11-11T12:00:00Z","by":"test"}`)); err != nil { + t.Fatal(err) + } + within(t, 5*time.Second, "the node did not answer the keeper's ask with a statement naming the finding", func() bool { + got, ok := lastEvent("findings-stated") + return ok && strings.Contains(got, `"asked":"a1"`) && strings.Contains(got, `"kind":"seat-list-unread"`) + }) 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. @@ -1048,6 +1077,15 @@ func TestNatsANodeThatCannotReadTheSeatListActsOnNothingNew(t *testing.T) { _, err := b.findings.Get(finding) return errors.Is(err, nats.ErrKeyNotFound) }) + within(t, 5*time.Second, "the node did not say finding-cleared for the seat list", func() bool { + for i := uint64(1); i < 40; i++ { + if m, err := b.js.GetMsg("MESH-HOST-FINDINGS", i); err == nil && m.Subject == findingSubject("finding-cleared", itNode) && + strings.Contains(string(m.Data), `"kind":"seat-list-unread"`) { + return true + } + } + return false + }) // Started again with the seat refusing: nothing new is composed, and what ran is neither removed nor changed. stop() diff --git a/cmd/mesh-host/seatlist.go b/cmd/mesh-host/seatlist.go index dd07ddb6..c53d423c 100644 --- a/cmd/mesh-host/seatlist.go +++ b/cmd/mesh-host/seatlist.go @@ -5,7 +5,10 @@ import ( "encoding/json" "errors" "fmt" + "slices" + "sort" "strings" + "sync" "sync/atomic" "time" @@ -103,29 +106,80 @@ 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 } +// findingsModule is the module whose own stream the node-engine says its findings on, and whose name its findings carry as +// their observer (module.json's `module` and its stream `mesh-host-findings`, design 47 G §383). +const findingsModule = "mesh-host" -func (w findingsOfTheNode) encode(f views.Finding) (string, []byte, error) { - // The seat list's finding is of the machine; a held module's is of the module on a machine, `.`. - scope, machine, source := conditions.ScopeMachine, f.ID, "seat-list-unread" +// 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. +type findingsOfTheNode struct { + session *link.Session + node string + + mu sync.Mutex + ctx context.Context + writer *views.Writer +} + +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, + Logf: func(format string, args ...any) { say(fmt.Sprintf(format, args...)) }, + Publish: w.session.PublishEvent, + Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) { + return w.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, source = conditions.ScopeModule, seatwait.HeldKind + scope = conditions.ScopeModule machine = f.ID[strings.LastIndex(f.ID, ".")+1:] } - key := conditions.Key(scope, f.ID, f.Kind) - v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning, - Subject: condseat.Subject{Scope: scope, ID: f.ID, Machine: machine}, Summary: f.Summary, - Headline: f.Headline, Explanation: f.Explanation, Source: source, At: f.At.UTC()} - if err := conditions.CheckFinding(v); err != nil { - return "", nil, err - } - raw, err := json.Marshal(v) - return key, raw, err + 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) { +func (w *findingsOfTheNode) left(node string) ([]views.Finding, error) { keys, err := w.session.FindingKeys() if err != nil { return nil, err @@ -139,37 +193,96 @@ func (w findingsOfTheNode) left(node string) ([]views.Finding, error) { if id, ok = strings.CutSuffix(id, "."+seatwait.HeldKind); !ok || !strings.HasSuffix(id, "."+node) { continue } - out = append(out, views.Finding{Observer: "node-engine", Kind: seatwait.HeldKind, ID: id}) + out = append(out, views.Finding{Observer: findingsModule, Kind: seatwait.HeldKind, ID: id}) } return out, nil } -func (w findingsOfTheNode) Put(f views.Finding) error { - key, raw, err := w.encode(f) +func (w *findingsOfTheNode) Put(f views.Finding) error { + ctx, writer, err := w.started() if err != nil { return err } - return w.session.PutFinding(key, raw) + 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 { - return conditions.Key(conditions.ScopeMachine, f.ID, f.Kind) +func (w *findingsOfTheNode) keyOf(f views.Finding) string { + b := findingBody(f) + return b.Key() } -func (w findingsOfTheNode) Delete(f views.Finding) error { - return w.session.DeleteFinding(w.keyOf(f)) +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) { +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 +} + // 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, say func(string)) (*seatListState, error) { +func startSeatList(ctx context.Context, session *link.Session, node string, findings seatwait.Findings, 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()] @@ -187,7 +300,7 @@ func startSeatList(ctx context.Context, session *link.Session, node string, say 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}, + st.waiter = &seatwait.Waiter{Node: node, Observer: findingsModule, Reader: reader, Findings: findings, 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 3e76ac4c..a9a8cb18 100644 --- a/cmd/mesh-host/seatlist_unread_test.go +++ b/cmd/mesh-host/seatlist_unread_test.go @@ -22,9 +22,9 @@ func TestAReaderThatCouldNotStartHoldsTheComposer(t *testing.T) { 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). - key, raw, err := findingsOfTheNode{}.encode(views.Finding{Observer: "node-engine", Kind: seatwait.HeldKind, + 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 err != nil || key != "module.mesh-grants.ace.module-held" || len(raw) == 0 { - t.Errorf("a held module's finding encodes as %q, %v", key, err) + 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 a4a46390..ab636c1a 100644 --- a/internal/link/findings.go +++ b/internal/link/findings.go @@ -1,9 +1,13 @@ package link import ( + "context" + "encoding/json" "errors" "fmt" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" ) @@ -89,3 +93,114 @@ func (o OverNATS) findingsKV() (nats.KeyValue, error) { } return kv, nil } + +// The node-engine's finding writer (mesh-sdk views.Writer, design 47 G §383, §384, §461) is made over the link open now: +// its state is the bucket above, its events are published on the engine's own stream awaiting the bus's acknowledgement, and +// its consumer of `findings-asked` is its own, from the next ask on. + +// FindingsState is the node-engine's state `findings` over the link open now, as the finding writer reads it +// (views.FindingState). +type FindingsState struct{ Session *Session } + +// Get is the key's entry, or none for a key the state does not hold. +func (s FindingsState) Get(key string) (*stdio.StateEntry, error) { + kv, err := s.Session.findings() + if err != nil { + return nil, err + } + e, err := kv.Get(key) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil +} + +// Put writes the value as JSON under the key. +func (s FindingsState) Put(key string, value any) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + kv, err := s.Session.findings() + if err != nil { + return 0, err + } + return kv.Put(key, raw) +} + +// Delete deletes the key; a key already gone is deleted. +func (s FindingsState) Delete(key string) error { return s.Session.DeleteFinding(key) } + +// Keys lists the keys the state holds. +func (s FindingsState) Keys() ([]string, error) { return s.Session.FindingKeys() } + +var _ views.FindingState = FindingsState{} + +// eventsOver is a link that publishes on a stream and makes a consumer of its own on another. +type eventsOver interface { + publishEvent(ctx context.Context, subject string, data []byte, id string) error + listenFrom(stream, subject string, handle func([]byte)) (func(), error) +} + +func publishEvent(js nats.JetStreamContext, ctx context.Context, subject string, data []byte, id string) error { + _, err := js.Publish(subject, data, nats.MsgId(id), nats.Context(ctx)) + return err +} + +func listenFrom(js nats.JetStreamContext, stream, subject string, handle func([]byte)) (func(), error) { + sub, err := js.Subscribe(subject, func(m *nats.Msg) { handle(m.Data) }, nats.BindStream(stream), nats.DeliverNew(), nats.AckNone()) + if err != nil { + return nil, fmt.Errorf("a consumer of %s: %w", subject, err) + } + return func() { _ = sub.Unsubscribe() }, nil +} + +func (l *natsLink) publishEvent(ctx context.Context, subject string, data []byte, id string) error { + return publishEvent(l.js, ctx, subject, data, id) +} + +func (l *natsLink) listenFrom(stream, subject string, handle func([]byte)) (func(), error) { + return listenFrom(l.js, stream, subject, handle) +} + +func (o OverNATS) publishEvent(ctx context.Context, subject string, data []byte, id string) error { + return publishEvent(o.JS, ctx, subject, data, id) +} + +func (o OverNATS) listenFrom(stream, subject string, handle func([]byte)) (func(), error) { + return listenFrom(o.JS, stream, subject, handle) +} + +func (q *Session) events() (eventsOver, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + e, ok := bus.(eventsOver) + if bus == nil || !ok { + return nil, ErrNoLink + } + return e, nil +} + +// PublishEvent publishes one message awaiting the bus's acknowledgement, over the link open now (views.WriterConfig.Publish). +func (q *Session) PublishEvent(ctx context.Context, subject string, data []byte, id string) error { + e, err := q.events() + if err != nil { + return err + } + return e.publishEvent(ctx, subject, data, id) +} + +// ListenFrom makes a consumer of its own on stream for subject that delivers only the messages published after it is +// made, over the link open now, and answers how to end it (views.WriterConfig.Listen). +func (q *Session) ListenFrom(stream, subject string, handle func([]byte)) (func(), error) { + e, err := q.events() + if err != nil { + return nil, err + } + return e.listenFrom(stream, subject, handle) +} diff --git a/internal/nodegrant/testdata/node.json b/internal/nodegrant/testdata/node.json index eded4963..6e82524d 100644 --- a/internal/nodegrant/testdata/node.json +++ b/internal/nodegrant/testdata/node.json @@ -3,8 +3,8 @@ "$JS.ACK.NODES.NODE.>", "$JS.API.CONSUMER.CREATE.KV_composed-NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-config_resolved.*.$KV.mesh-config_resolved.NODE.>", - "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_candidates.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_assignments.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_candidates.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_wants.*.$KV.mesh-delivery-assigner_wants.want.NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-builder_builds.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_candidates.>", @@ -13,17 +13,19 @@ "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_verdicts.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_acts.*.$KV.mesh-host_acts.NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_data.*.$KV.mesh-host_data.NODE.>", - "$JS.API.CONSUMER.CREATE.KV_mesh-host_findings.*.$KV.mesh-host_findings.machine.NODE.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-host_findings.*.$KV.mesh-host_findings.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_health.*.$KV.mesh-host_health.NODE", "$JS.API.CONSUMER.CREATE.KV_mesh-host_packages.*.$KV.mesh-host_packages.NODE", "$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_provisions.*.$KV.mesh-provisioner_provisions.NODE.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.*.node", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.NODE.>", "$JS.API.CONSUMER.CREATE.OBJ_declaration-files-NODE.*.$O.declaration-files-NODE.C.*", + "$JS.API.CONSUMER.CREATE.SEAT_mesh-conditions_findings-asked", + "$JS.API.CONSUMER.CREATE.SEAT_mesh-conditions_findings-asked.>", "$JS.API.CONSUMER.DELETE.KV_composed-NODE.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", - "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_candidates.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_assignments.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_candidates.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_wants.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-builder_builds.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_candidates.>", @@ -38,11 +40,12 @@ "$JS.API.CONSUMER.DELETE.KV_mesh-provisioner_provisions.>", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files-NODE.*", + "$JS.API.CONSUMER.DELETE.SEAT_mesh-conditions_findings-asked.>", "$JS.API.CONSUMER.INFO.NODES.NODE", "$JS.API.DIRECT.GET.KV_composed-NODE.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.NODE.>", - "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_candidates.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_assignments.>", + "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_candidates.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_wants.$KV.mesh-delivery-assigner_wants.want.NODE.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-builder_builds.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-indexer_candidates.>", @@ -51,7 +54,7 @@ "$JS.API.DIRECT.GET.KV_mesh-delivery-indexer_verdicts.>", "$JS.API.DIRECT.GET.KV_mesh-host_acts.$KV.mesh-host_acts.NODE.>", "$JS.API.DIRECT.GET.KV_mesh-host_data.$KV.mesh-host_data.NODE.>", - "$JS.API.DIRECT.GET.KV_mesh-host_findings.$KV.mesh-host_findings.machine.NODE.>", + "$JS.API.DIRECT.GET.KV_mesh-host_findings.$KV.mesh-host_findings.>", "$JS.API.DIRECT.GET.KV_mesh-host_health.$KV.mesh-host_health.NODE", "$JS.API.DIRECT.GET.KV_mesh-host_packages.$KV.mesh-host_packages.NODE", "$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.NODE.>", @@ -60,8 +63,8 @@ "$JS.API.DIRECT.GET.OBJ_declaration-files-NODE.$O.declaration-files-NODE.M.*", "$JS.API.STREAM.INFO.KV_composed-NODE", "$JS.API.STREAM.INFO.KV_mesh-config_resolved", - "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_candidates", "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_assignments", + "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_candidates", "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_wants", "$JS.API.STREAM.INFO.KV_mesh-delivery-builder_builds", "$JS.API.STREAM.INFO.KV_mesh-delivery-indexer_candidates", @@ -75,10 +78,11 @@ "$JS.API.STREAM.INFO.KV_mesh-host_packages", "$JS.API.STREAM.INFO.KV_mesh-provisioner_provisions", "$JS.API.STREAM.INFO.KV_node-engine_have", + "$JS.API.STREAM.INFO.SEAT_mesh-conditions_findings-asked", "$JS.FC.KV_composed-NODE.>", "$JS.FC.KV_mesh-config_resolved.>", - "$JS.FC.KV_mesh-delivery-assigner_candidates.>", "$JS.FC.KV_mesh-delivery-assigner_assignments.>", + "$JS.FC.KV_mesh-delivery-assigner_candidates.>", "$JS.FC.KV_mesh-delivery-assigner_wants.>", "$JS.FC.KV_mesh-delivery-builder_builds.>", "$JS.FC.KV_mesh-delivery-indexer_candidates.>", @@ -96,7 +100,7 @@ "$KV.composed-NODE.>", "$KV.mesh-host_acts.NODE.>", "$KV.mesh-host_data.NODE.>", - "$KV.mesh-host_findings.machine.NODE.>", + "$KV.mesh-host_findings.>", "$KV.mesh-host_health.NODE", "$KV.mesh-host_packages.NODE", "$KV.node-engine_have.NODE.>", @@ -111,7 +115,8 @@ "mesh.seat.node-backup.call.now.NODE.node~NODE", "mesh.seat.node-backup.tool.backed-up.NODE", "mesh.seat.node-backup.tool.now.NODE", - "mesh.seat.node-engine.event.*.NODE" + "mesh.seat.node-engine.event.*.NODE", + "mesh.stream.mesh-host.event.*.NODE" ], "PublishDeny": null, "Subscribe": [ diff --git a/internal/seatwait/seatwait.go b/internal/seatwait/seatwait.go index db6c1c25..54a8f4f9 100644 --- a/internal/seatwait/seatwait.go +++ b/internal/seatwait/seatwait.go @@ -66,6 +66,9 @@ func (w *Waiter) key() views.Finding { 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. +func Waits(st seatlist.Status) bool { return waiting(st) } + // 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() {