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 +}