From 54680238a0d5cf6a0a2b5e50f29a091ce4d28c3c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:42:43 +0100 Subject: [PATCH] Read memberships from the grants part's state, what the mesh holds from the indexer and placement, and give bundles newest, since and watch on logs, so the tool runner asks the controller for nothing it keeps (issue 1955, ADR 0571 clauses 1, 4, 5, 7) Pins mesh-sdk bb993dd. The assignments stream stays the membership's fallback until the lease write; a log another module keeps is read as .. --- node-tools/go.mod | 2 +- node-tools/go.sum | 2 + node-tools/internal/bus/bus.go | 153 +++++++++++-- node-tools/internal/bus/log.go | 214 ++++++++++++++---- node-tools/internal/bus/streams_test.go | 31 +++ node-tools/internal/console/address.go | 189 ++++++++++------ node-tools/internal/console/address_test.go | 52 ++--- node-tools/internal/console/console_test.go | 24 +- node-tools/internal/console/runtimes_test.go | 22 +- .../internal/launch/event_answer_test.go | 4 + node-tools/internal/launch/launch.go | 23 +- node-tools/internal/launch/provision_test.go | 4 + node-tools/internal/meshtest/meshtest.go | 75 ++++++ node-tools/internal/runtime/log_test.go | 70 +++++- node-tools/internal/runtime/runtime.go | 50 +++- node-tools/module.json | 5 +- 16 files changed, 716 insertions(+), 204 deletions(-) diff --git a/node-tools/go.mod b/node-tools/go.mod index 3c04a4f..d2c7b12 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -8,7 +8,7 @@ require ( ) require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect diff --git a/node-tools/go.sum b/node-tools/go.sum index 0c0c2bb..5a33f25 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -8,6 +8,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214150-692dbf9a1791 h1:HVKR7kEN git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214150-692dbf9a1791/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 h1:nP/jrjm2Qt1wdmd1Bc+E6c4+PobAzW5o64SPArE+jaw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 33dea0e..4051a80 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -10,6 +10,7 @@ package bus import ( + "context" "crypto/sha256" "crypto/tls" "crypto/x509" @@ -17,6 +18,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/buscompose" "log" "regexp" "strings" @@ -180,9 +182,12 @@ type Conn struct { cred Credential mu sync.Mutex issued map[string]*Membership // module → membership; present with nil = followed, none issued - onNew []*func(Membership) - subs []*nats.Subscription - Logf func(format string, args ...any) + // fromGrants is every module whose membership the grants part's state memberships has given (novox/hq ADR 0571 + // clause 7): its assignments stream copy is then a fallback no longer applied. + fromGrants map[string]bool + onNew []*func(Membership) + subs []*nats.Subscription + Logf func(format string, args ...any) // answering is every subject this runtime answers on, so a subscription the bus refused can be // asked for again (novox/hq issue 222). answering map[string]*answered @@ -326,7 +331,7 @@ func Connect(cred Credential) (*Conn, error) { return nil, err } c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred, - issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}} + issued: map[string]*Membership{}, fromGrants: map[string]bool{}, Logf: log.Printf, answering: map[string]*answered{}} c.Follow(cred.Module) return c, nil } @@ -346,7 +351,10 @@ func MembershipSubject(node, module string) string { } // Follow reads a module's membership on this machine once and follows it live, so its tools are -// served where the mesh issued them (ADR 0175). +// served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key +// `.` (novox/hq ADR 0571 clause 7, mesh-sdk buscompose.GrantsMembershipsBucket); the controller's +// assignments stream is read beside it as the fallback until the lease write, and applied only while the state has given +// the module no membership. func (c *Conn) Follow(module string) { c.mu.Lock() if c.node == "" { @@ -359,16 +367,21 @@ func (c *Conn) Follow(module string) { } c.issued[module] = nil c.mu.Unlock() + c.followGrants(module) subject := MembershipSubject(c.node, module) // The subject-addressed direct get: the one address the mesh grants this account on the // stream's API. - if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil { - if got.Header.Get("Status") == "" && len(got.Data) > 0 { - var m Membership - if json.Unmarshal(got.Data, &m) == nil { - c.mu.Lock() - c.issued[module] = &m - c.mu.Unlock() + if c.Membership(module) == nil { + if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil { + if got.Header.Get("Status") == "" && len(got.Data) > 0 { + var m Membership + if json.Unmarshal(got.Data, &m) == nil { + c.mu.Lock() + if !c.fromGrants[module] { + c.issued[module] = &m + } + c.mu.Unlock() + } } } } @@ -381,24 +394,116 @@ func (c *Conn) Follow(module string) { c.Logf("[mesh-tools] a membership arrived that is not one: %v", err) return } - c.mu.Lock() - c.issued[module] = &m - handlers := make([]func(Membership), 0, len(c.onNew)) - for _, h := range c.onNew { - handlers = append(handlers, *h) - } - c.mu.Unlock() - c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node) - for _, h := range handlers { - h(m) - } - _ = c.nc.Flush() + c.issue(module, m, false) }) if err == nil { c.track(sub) } } +// GrantsPatience bounds the first read of a membership from the grants part's state. +var GrantsPatience = 2 * time.Second + +// issue applies a membership that arrived for module, live, and re-serves on it. One from the assignments stream is +// passed over once the grants part's state has given the module one. +func (c *Conn) issue(module string, m Membership, fromGrants bool) { + c.mu.Lock() + if !fromGrants && c.fromGrants[module] { + c.mu.Unlock() + return + } + if fromGrants { + c.fromGrants[module] = true + } + c.issued[module] = &m + handlers := make([]func(Membership), 0, len(c.onNew)) + for _, h := range c.onNew { + handlers = append(handlers, *h) + } + c.mu.Unlock() + c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node) + for _, h := range handlers { + h(m) + } + _ = c.nc.Flush() +} + +// followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied +// before Follow returns, each change after it live. A state that is not there yet, or not readable, is said once, and +// the assignments stream answers alone. +func (c *Conn) followGrants(module string) { + key := buscompose.MembershipKey(c.node, module) + // First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet, + // answers at once, where binding the bucket would wait out its bound on a refusal. + bucket := buscompose.GrantsMembershipsBucket + probe, err := c.nc.Request("$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+key, nil, GrantsPatience) + if err != nil || probe.Header.Get("Status") == "503" { + why := "it did not answer" + if err != nil { + why = err.Error() + } + c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s's membership is read from the "+ + "assignments stream alone", bucket, why, module) + return + } + js, err := jetstream.New(c.nc) + if err != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := js.KeyValue(ctx, bucket) + if err != nil { + c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s's membership is read from the "+ + "assignments stream alone", buscompose.GrantsMembershipsBucket, err, module) + return + } + w, err := kv.Watch(context.Background(), key) + if err != nil { + c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); read from the "+ + "assignments stream alone", module, key, err) + return + } + apply := func(e jetstream.KeyValueEntry) { + if e.Operation() != jetstream.KeyValuePut { + return // a membership withdrawn is the stream's, or the module's unassignment's, to say + } + var m Membership + if err := json.Unmarshal(e.Value(), &m); err != nil { + c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err) + return + } + c.issue(module, m, true) + } + timer := time.NewTimer(StateTimeout) + defer timer.Stop() + for current := true; current; { + select { + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil { + current = false + continue + } + apply(e) + case <-timer.C: + _ = w.Stop() + c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; read from the "+ + "assignments stream alone", module, key, StateTimeout) + return + } + } + go func() { + for e := range w.Updates() { + if e != nil { + apply(e) + } + } + }() +} + // Membership is what the mesh issued a module here, or nil when nothing has been issued. func (c *Conn) Membership(module string) *Membership { c.mu.Lock() diff --git a/node-tools/internal/bus/log.go b/node-tools/internal/bus/log.go index 6ac1e0e..423d0c7 100644 --- a/node-tools/internal/bus/log.go +++ b/node-tools/internal/bus/log.go @@ -1,6 +1,7 @@ package bus import ( + "context" "encoding/json" "errors" "fmt" @@ -94,6 +95,10 @@ func logFilter(l LogIssued, pattern string) (string, error) { } parts := strings.Split(pattern, ".") for i, p := range parts { + if p == "**" && i == len(parts)-1 { + // `**` for the rest, as a state's key pattern and mesh-sdk's log reader say it. + parts[i], p = ">", ">" + } switch { case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p): default: @@ -104,7 +109,7 @@ func logFilter(l LogIssued, pattern string) (string, error) { if len(pattern) > LogKeyMax { return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax) } - return l.Subject + "." + pattern, nil + return l.Subject + "." + strings.Join(parts, "."), nil } // issuedLog is the log a module may reach by a name, or the reason it may not. @@ -126,8 +131,9 @@ func (c *Conn) issuedLog(module, name string) (LogIssued, error) { if len(names) > 0 { issued = strings.Join(names, ", ") } - return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, "+ - "and a log is its owner's alone; it was issued: %s (novox/hq ADR 0297)", module, name, issued) + return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, and "+ + "another module's it reads under `reads` as .; it was issued: %s (novox/hq ADR 0297, ADR 0571)", + module, name, issued) } // notOnTheBus says a log is issued and its stream is not there yet. @@ -200,9 +206,10 @@ func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect return ack.Seq, nil } -// LogRead is the entries from a sequence on, of the keys matching a pattern, at most limit of them and -// about 4 MiB, by one batch direct get. -func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) (*LogPage, error) { +// LogRead is the entries from a sequence on, or from a time on (since, when not zero), of the keys matching a pattern, at +// most limit of them and about 4 MiB, by one batch direct get. A log another module keeps is read as its own is, by the +// name its reads declare, `.` (novox/hq ADR 0571). +func (c *Conn) LogRead(module, name string, from uint64, since time.Time, keys string, limit int) (*LogPage, error) { l, err := c.issuedLog(module, name) if err != nil { return nil, err @@ -218,52 +225,24 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) if err != nil { return nil, err } + if from != 0 && !since.IsZero() { + return nil, fmt.Errorf("a read of %s starts from a sequence or from a time, not both", name) + } if from == 0 { from = 1 } - req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}) - inbox := c.nc.NewInbox() - sub, err := c.nc.SubscribeSync(inbox) + req := map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes} + if !since.IsZero() { + delete(req, "seq") + req["start_time"] = since.UTC() + } + entries, more, err := c.logBatch(l, req) if err != nil { return nil, err } - defer func() { _ = sub.Unsubscribe() }() - if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil { - return nil, err - } - page := &LogPage{Entries: []LogEntry{}} - deadline := time.Now().Add(StateTimeout) - for done := false; !done; { - msg, err := sub.NextMsg(time.Until(deadline)) - if err != nil { - if errors.Is(err, nats.ErrNoResponders) { - return nil, notOnTheBus(l) - } - if errors.Is(err, nats.ErrTimeout) { - return nil, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout) - } - return nil, err - } - switch status := msg.Header.Get("Status"); status { - case "": - e, err := logEntryOf(l, msg) - if err != nil { - return nil, err - } - page.Entries = append(page.Entries, *e) - case "204": // the end of the batch: what is still pending after the last entry sent - pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64) - if pending > 0 && len(page.Entries) > 0 { - page.Next = page.Entries[len(page.Entries)-1].Sequence + 1 - } - done = true - case "404": // nothing from that sequence on - done = true - case "503": - return nil, notOnTheBus(l) - default: - return nil, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description")) - } + page := &LogPage{Entries: entries} + if more && len(entries) > 0 { + page.Next = entries[len(entries)-1].Sequence + 1 } info, err := c.logInfo(l) if err != nil { @@ -273,6 +252,149 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) return page, nil } +// LogNewest is the newest entry of each key matching a pattern, in the log's order, by one batch direct get (mesh-sdk +// process.LogReader.Newest). +func (c *Conn) LogNewest(module, name, keys string) ([]LogEntry, error) { + l, err := c.issuedLog(module, name) + if err != nil { + return nil, err + } + filter, err := logFilter(l, keys) + if err != nil { + return nil, err + } + entries, more, err := c.logBatch(l, map[string]any{"multi_last": []string{filter}, "batch": LogReadMax}) + if err != nil { + return nil, err + } + if more { + return nil, fmt.Errorf("%s holds more than %d keys matching %q: name fewer", name, LogReadMax, keys) + } + return entries, nil +} + +// logBatch is one batch direct get of a log: its entries, and whether more are pending after them. +func (c *Conn) logBatch(l LogIssued, request map[string]any) ([]LogEntry, bool, error) { + req, _ := json.Marshal(request) + inbox := c.nc.NewInbox() + sub, err := c.nc.SubscribeSync(inbox) + if err != nil { + return nil, false, err + } + defer func() { _ = sub.Unsubscribe() }() + if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil { + return nil, false, err + } + entries := []LogEntry{} + deadline := time.Now().Add(StateTimeout) + for { + msg, err := sub.NextMsg(time.Until(deadline)) + if err != nil { + if errors.Is(err, nats.ErrNoResponders) { + return nil, false, notOnTheBus(l) + } + if errors.Is(err, nats.ErrTimeout) { + return nil, false, fmt.Errorf("the read of %s did not end in %s", l.Name, StateTimeout) + } + return nil, false, err + } + switch status := msg.Header.Get("Status"); status { + case "": + e, err := logEntryOf(l, msg) + if err != nil { + return nil, false, err + } + entries = append(entries, *e) + case "204": // the end of the batch: what is still pending after the last entry sent + pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64) + return entries, pending > 0, nil + case "404": // nothing from there on + return entries, false, nil + case "503": + return nil, false, notOnTheBus(l) + default: + return nil, false, fmt.Errorf("the read of %s was refused by the bus: %s %s", l.Name, status, msg.Header.Get("Description")) + } + } +} + +// LogWatch hands deliver every entry of the keys matching a pattern from a sequence on (from), from a time on (since), +// or, with neither, each new one, in the log's order, one at a time, through an ordered consumer it lets go of when +// stopped (mesh-sdk process.LogReader.Watch). It returns once the entries there at its start are handed over; an error +// from deliver is said, and the watch goes on with the next entry. +func (c *Conn) LogWatch(module, name, keys string, from uint64, since time.Time, deliver func(LogEntry) error) (func(), error) { + l, err := c.issuedLog(module, name) + if err != nil { + return nil, err + } + filter, err := logFilter(l, keys) + if err != nil { + return nil, err + } + cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverNewPolicy} + switch { + case from != 0 && !since.IsZero(): + return nil, fmt.Errorf("a watch of %s starts from a sequence or from a time, not both", name) + case from != 0: + cfg.DeliverPolicy, cfg.OptStartSeq = jetstream.DeliverByStartSequencePolicy, from + case !since.IsZero(): + at := since.UTC() + cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at + } + js, err := jetstream.New(c.nc) + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + cons, err := js.OrderedConsumer(ctx, l.Stream, cfg) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return nil, notOnTheBus(l) + } + if err != nil { + return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err) + } + it, err := cons.Messages() + if err != nil { + return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err) + } + hand := func(m jetstream.Msg) error { + meta, err := m.Metadata() + if err != nil { + return err + } + e := LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.Subject+"."), + Value: valueOf(m.Data()), At: meta.Timestamp.UTC()} + return deliver(e) + } + for pending := info.NumPending; pending > 0; pending-- { + m, err := it.Next() + if err == nil { + err = hand(m) + } + if err != nil { + it.Stop() + return nil, fmt.Errorf("the watch of %s: %w", name, err) + } + } + go func() { + for { + m, err := it.Next() + if err != nil { + return // stopped, or the connection closed + } + if err := hand(m); err != nil { + c.Logf("[mesh-tools] %s did not take entry of %s: %v; it is in the log, a read finds it", module, name, err) + } + } + }() + return it.Stop, nil +} + // LogLast is a key's last entry, or nil when it has none. func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) { l, err := c.issuedLog(module, name) diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go index da8bde3..4319662 100644 --- a/node-tools/internal/bus/streams_test.go +++ b/node-tools/internal/bus/streams_test.go @@ -160,6 +160,37 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) { } return nil }) + + // The membership as the grants part keeps it (ADR 0571 clause 7): read first, applied live, and the assignments + // stream's copy passed over once the state has given one. + granted := reader + granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}} + mesh.IssueInGrants(t, granted) + again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + again.Logf = func(string, ...any) {} + t.Cleanup(again.Close) + again.Follow("reader") + consumerOf := func() string { + if m := again.Membership("reader"); m != nil && len(m.Hears) == 1 { + return m.Hears[0].Consumer + } + return "" + } + if got := consumerOf(); got != "from-grants" { + t.Fatalf("the membership read at the follow names %q, want the grants part's", got) + } + mesh.Issue(t, reader) + granted.Hears[0].Consumer = "from-grants-2" + mesh.IssueInGrants(t, granted) + mt.Until(t, func() error { + if got := consumerOf(); got != "from-grants-2" { + return fmt.Errorf("the membership names %q, want the grants part's change", got) + } + return nil + }) } // novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index 2cf13ab..b8b25d1 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -131,7 +131,7 @@ type index struct { NotAnswering []string Listing *Listing // the flat catalogue the call path resolves subjects with Discovery announce.Discovery - // Recorded is where the controller's records place each module that declares tools. + // Recorded is where placement places each module that declares tools. Recorded map[string][]string // Unheard is every runtime whose answer this index lacks: it said it was there and did not say // what it serves in time, or it answered before and not now. An address it might answer is never @@ -231,42 +231,8 @@ func findTool(tools []Tool, name string) *Tool { return nil } -// controllerOutput is a controller seat verb's answer: the command's printed output. -func controllerOutput(conn *bus.Conn, verb string) (string, error) { - got, err := conn.Ask("seat:mesh-controller."+verb, map[string]any{}, "") - if err != nil { - return "", err - } - var r struct { - Output string `json:"output"` - OK *bool `json:"ok"` - } - if json.Unmarshal(got.Result, &r) != nil { - return "", fmt.Errorf("mesh-controller.%s answered something that is not its output", verb) - } - if r.OK != nil && !*r.OK { - return "", fmt.Errorf("mesh-controller.%s: %s", verb, strings.TrimSpace(r.Output)) - } - return r.Output, nil -} - -// jsonIn is the JSON document a command printed, after any lines it said first: a seat verb runs -// the controller's command, and a command may warn before it answers. -func jsonIn(output string) string { - t := strings.TrimSpace(output) - if strings.HasPrefix(t, "{") || strings.HasPrefix(t, "[") { - return t - } - for _, open := range []string{"\n{", "\n["} { - if i := strings.Index(output, open); i >= 0 { - return strings.TrimSpace(output[i+1:]) - } - } - return "" -} - -// recordedModule is a module as the controller's records hold it (`module list --json`): where the -// mesh assigned it, and whether it declares tools — what should announce itself, and where. +// recordedModule is a module as the bus holds it: where the placement seat places it, and whether its manifest declares +// tools — what should announce itself, and where (novox/hq ADR 0571). type recordedModule struct { Module string `json:"module"` On []string `json:"on"` @@ -275,11 +241,6 @@ type recordedModule struct { Replaces map[string][]string `json:"replaces,omitempty"` } -// recordedMachine is a machine as the controller's records hold it (`node list --json`). -type recordedMachine struct { - Name string `json:"name"` -} - // indexOn asks the mesh what it holds (novox/hq ADR 0197): what answers, from every runtime's own // announcement on the bus — one `$SRV.INFO` request — and what should, from the controller's records // read as JSON. Nothing is inferred from a roster and nothing is parsed from print. @@ -289,12 +250,15 @@ type recordedMachine struct { // index holds instead, marked with its age (novox/hq issue 554). func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) { var wg sync.WaitGroup - var nodesOut, modulesOut string var placed map[string][]byte var placedErr error - wg.Add(3) - go func() { defer wg.Done(); nodesOut, _ = controllerOutput(conn, "nodes") }() - go func() { defer wg.Done(); modulesOut, _ = controllerOutput(conn, "modules") }() + var recorded []recordedModule + var known []string + var recordsWhy []string + wg.Add(2) + // Which modules there are and where they are placed, and which machines the mesh has, from the bus (ADR 0571 + // clauses 4 and 5): the indexer's modules and manifests and the placement seat's placements, never the controller. + go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(conn) }() // Who holds which seat on which node, from placement (ADR 0571 clause 4): no call of the controller. go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(conn) }() d, err := announce.Gather(conn, held) @@ -346,26 +310,21 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery // What should have answered: every assignment of a module that declares tools. Silence is named; // a module with no tools is never a name here. - var recorded []recordedModule - recordedOK := json.Unmarshal([]byte(jsonIn(modulesOut)), &recorded) == nil x.Replaces = replacesOf(catalogue.DefaultSeats(), recorded) - if recordedOK { - for _, m := range recorded { - if !m.Tools { - continue - } - x.Recorded[m.Module] = append([]string{}, m.On...) - for _, n := range m.On { - // A holder of a seat held once for the mesh announces no machine — the seat is the mesh's, - // not a machine's — so what it announced without one answers for wherever it is assigned. - if !announced[m.Module][n] && !announced[m.Module][""] { - l.NotAnswering = append(l.NotAnswering, m.Module+" on "+n) - } + for _, m := range recorded { + if !m.Tools { + continue + } + x.Recorded[m.Module] = append([]string{}, m.On...) + for _, n := range m.On { + // A holder of a seat held once for the mesh announces no machine — the seat is the mesh's, + // not a machine's — so what it announced without one answers for wherever it is assigned. + if !announced[m.Module][n] && !announced[m.Module][""] { + l.NotAnswering = append(l.NotAnswering, m.Module+" on "+n) } } - } else { - l.NotAnswering = append(l.NotAnswering, "mesh-controller (its records of the modules did not answer, so what is missing cannot be said)") } + l.NotAnswering = append(l.NotAnswering, recordsWhy...) if x.PlacedWhy != "" { l.NotAnswering = append(l.NotAnswering, delivery.SeatHoldersRead+" (who holds which seat by placement could not be read: "+ x.PlacedWhy+")") @@ -376,13 +335,8 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery sort.Strings(l.NotAnswering) x.NotAnswering = l.NotAnswering - var known []recordedMachine - if json.Unmarshal([]byte(jsonIn(nodesOut)), &known) == nil { - for _, n := range known { - if n.Name != "" { - machines[n.Name] = true - } - } + for _, n := range known { + machines[n] = true } for n := range machines { x.Machines = append(x.Machines, n) @@ -391,12 +345,11 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery return x, nil } -// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4): its current values, through the -// runtime's own membership, which reads it as delivery.SeatHoldersRead. -func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) { +// stateOn reads a state the runtime's membership reads, whole: its current values by key, a deleted key absent. +func stateOn(conn *bus.Conn, name string) (map[string][]byte, error) { var mu sync.Mutex keys := map[string][]byte{} - stop, err := conn.StateWatch(conn.Credential().Module, delivery.SeatHoldersRead, "", func(c bus.StateChange) error { + stop, err := conn.StateWatch(conn.Credential().Module, name, "", func(c bus.StateChange) error { if c.Current && c.Op == "put" { mu.Lock() keys[c.Key] = c.Value @@ -413,6 +366,96 @@ func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) { return keys, nil } +// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as delivery.SeatHoldersRead. +func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) { + return stateOn(conn, delivery.SeatHoldersRead) +} + +// The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its +// digest, by the names a reader's manifest declares. +var ( + modulesRead = delivery.IndexerModule + ".modules" + manifestsRead = delivery.IndexerModule + ".manifests" +) + +// manifestFacts is what the console takes of a manifest: the tools it declares and what each replaces. A manifest's +// digest never names another, so what is read once is kept. +type manifestFacts struct { + Tools []string `json:"tools"` + Replaces map[string][]string `json:"replaces"` +} + +var manifestsKept sync.Map // digest → manifestFacts + +// recordsOn is every module the indexer holds, with where the placement seat places it and whether it declares tools, +// and every machine placement names (delivery.ModuleNames, delivery.MachineNames). What cannot be read is said, one line +// per state, for the listing's not answering. +func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, why []string) { + placements, err := stateOn(conn, delivery.PlacementRead) + if err != nil { + why = append(why, delivery.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+ + "be said: "+err.Error()+")") + } + machines = delivery.MachineNames(placements) + on := map[string][]string{} + for _, node := range machines { + key, _ := delivery.PlacementKey(node) + var p delivery.Placement + if json.Unmarshal(placements[key], &p) != nil { + why = append(why, delivery.PlacementRead+" ("+key+" is not a placement)") + continue + } + for _, m := range p.Modules { + on[m.Module] = append(on[m.Module], node) + } + } + modules, err := stateOn(conn, modulesRead) + if err != nil { + return nil, machines, append(why, modulesRead+" (which modules there are could not be read: "+err.Error()+")") + } + names, err := delivery.ModuleNames(modules) + if err != nil { + return nil, machines, append(why, modulesRead+" ("+err.Error()+")") + } + for _, name := range names { + r := recordedModule{Module: name, On: on[name]} + if m, err := delivery.DecodeModule(modules[name]); err == nil && m.Manifest != "" { + facts, err := manifestOf(conn, m.Manifest) + if err != nil { + why = append(why, name+" (its manifest could not be read, so whether it answers tools is not known: "+err.Error()+")") + } else { + r.Tools, r.Replaces = len(facts.Tools) > 0, facts.Replaces + } + } + recorded = append(recorded, r) + } + return recorded, machines, why +} + +// manifestOf is what the console takes of the manifest of a digest, read from the indexer's manifests once. +func manifestOf(conn *bus.Conn, digest string) (manifestFacts, error) { + if kept, ok := manifestsKept.Load(digest); ok { + return kept.(manifestFacts), nil + } + entry, err := conn.StateGet(conn.Credential().Module, manifestsRead, digest) + if err != nil { + return manifestFacts{}, err + } + if entry == nil { + return manifestFacts{}, fmt.Errorf("the indexer holds no manifest %s", digest) + } + m, err := delivery.DecodeManifest(entry.Value) + if err != nil { + return manifestFacts{}, err + } + var facts manifestFacts + if err := json.Unmarshal([]byte(m.JSON), &facts); err != nil { + return manifestFacts{}, err + } + manifestsKept.Store(digest, facts) + return facts, nil +} + // placedHolders is who holds seat (on node, for a seat held per machine) by placement, or none. func (x *index) placedHolders(seat, node string) []delivery.SeatHolder { if len(x.Placed) == 0 { diff --git a/node-tools/internal/console/address_test.go b/node-tools/internal/console/address_test.go index 6342941..883589e 100644 --- a/node-tools/internal/console/address_test.go +++ b/node-tools/internal/console/address_test.go @@ -70,29 +70,28 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { t.Cleanup(stop) } - // The controller: its records as JSON, as its seat verbs answer them, and its own seat's verbs - // announced on the bus like every runtime's. + // The controller: its own seat's verbs announced on the bus like every runtime's; its records are asked by no one. controller := connect(t, "mesh-controller", "bench") - out := func(s string) map[string]any { return map[string]any{"output": s, "ok": true} } - serve := func(verb string, answer func() any) { - stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) { - return answer(), nil - }) - if err != nil { - t.Fatal(err) - } - t.Cleanup(stop) - } - serve("nodes", func() any { - return out("the bus's user list leaves out 2 user(s)\n" + - `[{"name":"bench","heard":"3m ago","mode":"converged","id":"1f2e"},{"name":"desk","heard":"here","mode":"converged","id":"9a8b"}]`) + // Its verb nodes, called through the console as any seat's verb is. + stopNodes, err := controller.HandleSubject("mesh.seat.mesh-controller.tool.nodes", func(json.RawMessage) (any, error) { + return map[string]any{"ok": true, "output": `[{"name":"bench","mode":"converged"},{"name":"desk","mode":"converged"}]`}, nil }) - var gammaOn atomic.Value - gammaOn.Store("[]") - serve("modules", func() any { - return out(fmt.Sprintf(`[{"module":"alpha","on":["desk"],"tools":true},{"module":"beta","on":["desk"],"tools":true},`+ - `{"module":"gamma","on":%s,"tools":true},{"module":"delta","on":["desk"],"tools":false},`+ - `{"module":"epsilon","on":["bench"],"tools":true},{"module":"mesh-controller","on":["bench"],"tools":true}]`, gammaOn.Load().(string))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stopNodes) + // What the mesh holds, as the indexer and placement write it (ADR 0571 clauses 4, 5): never the controller's records. + tools := map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "gamma": {"given"}, "delta": nil, + "epsilon": {"seven"}, "mesh-controller": {"nodes"}} + mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta"}, "bench": {"epsilon", "mesh-controller"}}) + console := mt.MembershipOf("node-tools", "desk", false, nil) + console.State = mt.ConsoleReads() + mesh.Issue(t, console) + mt.Until(t, func() error { + if m := nodeTools.Membership("node-tools"); m == nil || len(m.State) == 0 { + return fmt.Errorf("the console's reads are not issued yet") + } + return nil }) stopAnn, err := announce.Serve(controller, announce.Service{Name: "mesh-controller", ID: "bench"}, func() []announce.Endpoint { return []announce.Endpoint{{Kind: announce.KindSeat, Module: "mesh-controller", Seat: "mesh-controller", Scope: "mesh", @@ -202,7 +201,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { t.Fatalf("gamma was found before it served: %s", got) } mesh.Issue(t, mt.MembershipOf("gamma", "desk", false, nil)) - gammaOn.Store(`["desk"]`) + mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta", "gamma"}, "bench": {"epsilon", "mesh-controller"}}) late := connect(t, "node-tools", "desk") stopLate, err := runtime.Run(late, []runtime.Served{{Module: "gamma", Entrypoints: []string{mt.Fixture("env-gamma.serve.mjs")}}}, nil, (&mt.Logs{}).Logf) @@ -233,12 +232,3 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { t.Errorf("an old name: %s", got) } } - -func TestTheControllersRecordsAreReadAsJSON(t *testing.T) { - if got := jsonIn("a warning printed first\n[{\"name\":\"ace\"}]"); got != `[{"name":"ace"}]` { - t.Errorf("an array after a warning: %q", got) - } - if got := jsonIn(`{"seats":[]}`); got != `{"seats":[]}` { - t.Errorf("an object: %q", got) - } -} diff --git a/node-tools/internal/console/console_test.go b/node-tools/internal/console/console_test.go index 11fbf5c..2ad8e70 100644 --- a/node-tools/internal/console/console_test.go +++ b/node-tools/internal/console/console_test.go @@ -3,6 +3,7 @@ package console import ( "bytes" "encoding/json" + "fmt" "io" "net/http" "strings" @@ -56,14 +57,19 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) { } defer stop() - // The controller's records (ADR 0197): ghost declares tools on desk and nothing announces it. - controller := connect(t, "mesh-controller", "") - stopSeat, _ := controller.HandleSubject("mesh.seat.mesh-controller.tool.modules", func(json.RawMessage) (any, error) { - return map[string]any{"ok": true, "output": `[{"module":"alpha","on":["desk"],"tools":true},` + - `{"module":"beta","on":["desk"],"tools":true},{"module":"ghost","on":["desk"],"tools":true}]`}, nil + // What the mesh holds, as the indexer and placement write it (ADR 0571): ghost declares tools on desk and nothing + // announces it. + mesh.Placed(t, map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "ghost": {"boo"}}, + map[string][]string{"desk": {"alpha", "beta", "ghost"}}) + console := mt.MembershipOf("node-tools", "desk", false, nil) + console.State = mt.ConsoleReads() + mesh.Issue(t, console) + mt.Until(t, func() error { + if m := nodeTools.Membership("node-tools"); m == nil || len(m.State) == 0 { + return fmt.Errorf("the console's reads are not issued yet") + } + return nil }) - defer stopSeat() - controller.Flush() flat := NewSurface(nodeTools, "desk.node-tools") flat.Flat = true // the whole catalogue, as before ADR 0195: still reachable, no longer announced @@ -86,9 +92,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) { if got := strings.Join(names, ","); got != "alpha.one,alpha.two,beta.five,beta.four,beta.three,node-shelf.clear,node-shelf.list" { t.Errorf("listed %s", got) } - // The console's runtime is issued no read of seat-holders here: that is said beside ghost, never dropped (ADR 0571). - if got := listed["_meta"].(map[string]any)["notAnswering"]; len(got.([]any)) != 2 || got.([]any)[0] != "ghost on desk" || - !strings.HasPrefix(got.([]any)[1].(string), "mesh-delivery-assigner.seat-holders (who holds which seat by placement could not be read") { + if got := listed["_meta"].(map[string]any)["notAnswering"]; len(got.([]any)) != 1 || got.([]any)[0] != "ghost on desk" { t.Errorf("not answering: %v", got) } for _, x := range listed["tools"].([]any) { diff --git a/node-tools/internal/console/runtimes_test.go b/node-tools/internal/console/runtimes_test.go index ad543fa..720e271 100644 --- a/node-tools/internal/console/runtimes_test.go +++ b/node-tools/internal/console/runtimes_test.go @@ -56,6 +56,10 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) IndexKept, announce.Patience = time.Minute, 2500*time.Millisecond t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas }) mesh := mt.New(t) + // What the mesh holds, as the indexer and placement write it (ADR 0571): never the controller's records. + mesh.Placed(t, map[string][]string{"slack": {"slack_check"}, "nap": {"doze"}, "nextcloud-client": {"nextcloud_check"}}, + map[string][]string{"laptop": {"slack", "nextcloud-client"}, "sleeper": {"nap", "nextcloud-client"}, + "desktop": {"nextcloud-client"}, "bench": nil}) // Who holds which seat, by placement (ADR 0571 clause 4), as the seat mesh-placement's holder composes it. placedKV := mesh.Bucket(t, delivery.SeatHoldersBucket) for key, sh := range delivery.SeatHoldersOf(map[string]delivery.Placement{"bench": {Node: "bench", @@ -66,7 +70,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) } } consoleMembership := mt.MembershipOf("node-tools", "laptop-console", false, nil) - consoleMembership.State = []bus.StateIssued{{Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket}} + consoleMembership.State = mt.ConsoleReads() mesh.Issue(t, consoleMembership) laptop := connect(t, "node-tools", "laptop") @@ -94,18 +98,14 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) controller := connect(t, "mesh-controller", "bench") var toolsAsked atomic.Int32 for verb, output := range map[string]string{ - "nodes": `[{"name":"laptop"},{"name":"sleeper"},{"name":"desktop"},{"name":"bench"}]`, - "modules": `[{"module":"slack","on":["laptop"],"tools":true},{"module":"nap","on":["sleeper"],"tools":true},` + - `{"module":"nextcloud-client","on":["laptop","desktop","sleeper"],"tools":true}]`, - // No part keeps a central record of tools (ADR 0571 clause 5): a seat verb's `replaces` is read from its contract - // in mesh-sdk, so the controller's `tools` is served here only to see that it is never asked. - "tools": `{"seats":[]}`, + // The controller's records are served here only to see that none is asked (ADR 0571 clauses 4 and 5). + "nodes": `[{"name":"laptop"}]`, + "modules": `[]`, + "tools": `{"seats":[]}`, } { output, verb := output, verb stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) { - if verb == "tools" { - toolsAsked.Add(1) - } + toolsAsked.Add(1) return map[string]any{"output": output, "ok": true}, nil }) if err != nil { @@ -149,7 +149,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) t.Errorf("mesh/seats answered %s", raw) } if n := toolsAsked.Load(); n != 0 { - t.Errorf("the console asked the controller's tools %d times", n) + t.Errorf("the console asked the controller's records %d times", n) } // The silent one is named, never called missing. diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 05d8594..7ccfc47 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -95,3 +95,7 @@ func TestAnEventIsTakenByAnyAnswerThatIsNotAnError(t *testing.T) { t.Fatalf("an error answer was not the bundle's refusal of mesh/event: %#v", err) } } + +func (b *subscribing) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { + return func() {}, nil +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index f9960aa..43f3fff 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -80,6 +80,8 @@ type Bus interface { Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) // Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297). Log(verb string, params json.RawMessage) (json.RawMessage, error) + // LogWatch hands deliver each entry of a log from where the bundle asked, one at a time (novox/hq ADR 0571). + LogWatch(params json.RawMessage, deliver func(entry json.RawMessage) error) (stop func(), err error) // Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once; // `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child // of the module asked last — a restarted child asks again as its code runs again. @@ -328,7 +330,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil { result = answered } - case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info": + case "mesh/log.append", "mesh/log.read", "mesh/log.newest", "mesh/log.last", "mesh/log.info": var answered json.RawMessage if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil { result = answered @@ -357,6 +359,25 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an c.mu.Unlock() } } + case "mesh/log.watch": + // As a state's watch: each entry asked of this child in order, `mesh/log {watch, entry}`, + // naming the watch by the id the child asked it with (novox/hq ADR 0571). + var stop func() + watch := m.ID + stop, err = mesh.LogWatch(m.Params, func(entry json.RawMessage) error { + _, err := c.ask(module, "mesh/log", map[string]any{"watch": watch, "entry": entry}, EventTimeout) + return err + }) + if err == nil { + c.mu.Lock() + if c.why != nil { + c.mu.Unlock() + stop() + } else { + c.watches = append(c.watches, stop) + c.mu.Unlock() + } + } case "mesh/state.watch": // Each change is asked of this child, in order, naming the watch it is for by // the id the child asked it with — two watches of one state are two handlers; diff --git a/node-tools/internal/launch/provision_test.go b/node-tools/internal/launch/provision_test.go index 44533ec..0b38435 100644 --- a/node-tools/internal/launch/provision_test.go +++ b/node-tools/internal/launch/provision_test.go @@ -105,3 +105,7 @@ func TestABundleAsksForItsValuesAndIsHandedItsWatchsChanges(t *testing.T) { t.Errorf("the runtime's bus was asked %q", b.asked) } } + +func (b *provisioning) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { + return func() {}, nil +} diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 09c1bb5..1673a0d 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -13,6 +13,8 @@ package meshtest import ( "context" "encoding/json" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/delivery" "os" "path/filepath" "runtime" @@ -106,6 +108,79 @@ func (m *Mesh) Issue(t *testing.T, mem bus.Membership) { } } +// IssueInGrants writes a membership where the grants part keeps it, its state memberships, key `.` +// (novox/hq ADR 0571 clause 7; mesh-sdk buscompose.GrantsMembershipsBucket), making the bucket when it is not there. +func (m *Mesh) IssueInGrants(t *testing.T, mem bus.Membership) { + t.Helper() + kv, err := m.js.KeyValue(buscompose.GrantsMembershipsBucket) + if err != nil { + if kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil { + t.Fatal(err) + } + } + body, _ := json.Marshal(mem) + if _, err := kv.Put(buscompose.MembershipKey(mem.Node, mem.Module), body); err != nil { + t.Fatal(err) + } +} + +// ConsoleReads are the states a console's runtime reads to say what the mesh holds (novox/hq ADR 0571 clauses 4, 5): +// placement, the indexer's modules and manifests, and the seat holders. +func ConsoleReads() []bus.StateIssued { + return []bus.StateIssued{ + {Name: delivery.PlacementRead, Bucket: delivery.PlacementBucket}, + {Name: delivery.IndexerModule + ".modules", Bucket: delivery.ModulesBucket}, + {Name: delivery.IndexerModule + ".manifests", Bucket: delivery.ManifestsBucket}, + {Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket}, + } +} + +// Placed writes what the mesh holds as its parts write it, with mesh-sdk's encoders: each module the indexer holds with +// its manifest (tools, the tools it declares; none for a module without), and each node's placement of modules. +func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]string) { + t.Helper() + kv := func(bucket string) nats.KeyValue { + _ = m.js.DeleteKeyValue(bucket) + b, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 1, MaxValueSize: 256 * 1024}) + if err != nil { + t.Fatal(err) + } + return b + } + modules, manifests, placement := kv(delivery.ModulesBucket), kv(delivery.ManifestsBucket), kv(delivery.PlacementBucket) + kv(delivery.SeatHoldersBucket) + for module, declared := range tools { + manifest, _ := json.Marshal(map[string]any{"module": module, "tools": declared}) + digest := delivery.ManifestDigest(manifest) + raw, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: string(manifest)}) + if err != nil { + t.Fatal(err) + } + if _, err := manifests.Put(digest, raw); err != nil { + t.Fatal(err) + } + raw, err = delivery.EncodeModule(delivery.Module{None: "not built here", Manifest: digest, Repository: "novox/" + module, + Commit: strings.Repeat("a", 40)}) + if err != nil { + t.Fatal(err) + } + if _, err := modules.Put(module, raw); err != nil { + t.Fatal(err) + } + } + for node, placed := range on { + p := delivery.Placement{V: delivery.Version, Node: node} + for _, module := range placed { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: module}) + } + raw, _ := json.Marshal(p) + key, _ := delivery.PlacementKey(node) + if _, err := placement.Put(key, raw); err != nil { + t.Fatal(err) + } + } +} + // NextEvent is the subject the next event under a pattern lands on. func (m *Mesh) NextEvent(t *testing.T, pattern string) <-chan string { t.Helper() diff --git a/node-tools/internal/runtime/log_test.go b/node-tools/internal/runtime/log_test.go index f664783..61a27af 100644 --- a/node-tools/internal/runtime/log_test.go +++ b/node-tools/internal/runtime/log_test.go @@ -20,9 +20,13 @@ func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt. t.Helper() mesh = mt.New(t) m := mt.MembershipOf("issues", "anchor", false, nil) - m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)} + changes := mesh.Log(t, "issues", "changes", maxBytes) + m.Logs = []bus.LogIssued{changes} mesh.Issue(t, m) - mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil)) + // other reads issues' log, as its manifest's reads declare it (ADR 0571): `.`, without writing it. + o := mt.MembershipOf("other", "anchor", false, nil) + o.Logs = []bus.LogIssued{{Name: "issues.changes", Stream: changes.Stream, Subject: changes.Subject}} + mesh.Issue(t, o) conn := connect(t, "node-tools", "anchor") conn.Follow("issues") conn.Follow("other") @@ -111,7 +115,7 @@ func infoOf(t *testing.T, b launch.Bus) map[string]uint64 { // RFC 3339; last is a key's last entry or null; info says entries, bytes, cap, first and last — and no // consumer is made for any of it. func TestALogAnswersEachVerbOnTheBus(t *testing.T) { - b, _, mesh := logBuses(t, 1<<20) + b, other, mesh := logBuses(t, 1<<20) before := time.Now().Add(-time.Second) var seqs []uint64 @@ -189,6 +193,64 @@ func TestALogAnswersEachVerbOnTheBus(t *testing.T) { if n := mesh.Consumers(t, "LOG_issues_changes"); n != 0 { t.Fatalf("the runtime made %d consumers of the log; it uses none", n) } + + // From a time (ADR 0571): `since` in place of `from`, RFC 3339; both at once are refused. + cut := all.Entries[2].At + if p := readOf(t, b, fmt.Sprintf(`{"log":"changes","since":%q}`, cut)); len(p.Entries) == 0 || p.Entries[0].Sequence > seqs[2] || + p.Entries[len(p.Entries)-1].Sequence != seqs[4] { + t.Fatalf("a read since %s: %+v", cut, p) + } + if p := readOf(t, b, fmt.Sprintf(`{"log":"changes","since":%q}`, time.Now().Add(time.Hour).Format(time.RFC3339))); len(p.Entries) != 0 { + t.Fatalf("a read since an hour ahead: %+v", p) + } + if _, err := logAsk(b, "read", `{"log":"changes","from":2,"since":"2026-10-29T00:00:00Z"}`); err == nil { + t.Fatal("a read from a sequence and a time was not refused") + } + // The newest of each key, in the log's order; `**` for the rest, as mesh-sdk says it. + raw, err = logAsk(b, "newest", `{"log":"changes"}`) + var newest []struct { + Sequence uint64 `json:"sequence"` + Key string `json:"key"` + } + if err != nil || json.Unmarshal(raw, &newest) != nil || len(newest) != 3 || newest[0].Key != "13" || newest[1].Key != "14.notes" || + newest[2].Sequence != seqs[4] { + t.Fatalf("newest: %s %v", raw, err) + } + if raw, err := logAsk(b, "newest", `{"log":"changes","keys":"14.**"}`); err != nil || !strings.Contains(string(raw), `"14.notes"`) || + strings.Contains(string(raw), `"12"`) { + t.Fatalf("newest of 14.**: %s %v", raw, err) + } + // Another module's log, read by the name its reads declare; never written. + if p := readOf(t, other, `{"log":"issues.changes","from":1}`); len(p.Entries) != 5 { + t.Fatalf("other's read of issues.changes: %+v", p) + } + if _, err := logAsk(other, "append", `{"log":"issues.changes","key":"1","value":{}}`); err == nil || + !strings.Contains(err.Error(), "without writing it") { + t.Fatalf("other's append to the log it reads: %v", err) + } + // A watch hands each entry from where it is asked, then each new one, one at a time. + heard := make(chan bus.LogEntry, 16) + stop, err := b.LogWatch(json.RawMessage(fmt.Sprintf(`{"log":"changes","keys":"12","from":%d}`, seqs[2])), func(raw json.RawMessage) error { + var e bus.LogEntry + _ = json.Unmarshal(raw, &e) + heard <- e + return nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + next := appendOf(t, b, `{"log":"changes","key":"12","value":{"op":"late"}}`) + for _, want := range []uint64{seqs[2], seqs[4], next} { + select { + case e := <-heard: + if e.Sequence != want || e.Key != "12" { + t.Fatalf("the watch handed %d %s, want %d", e.Sequence, e.Key, want) + } + case <-time.After(5 * time.Second): + t.Fatalf("the watch did not hand entry %d", want) + } + } } // A page holds about 4 MiB at most, whatever its limit, and says where to continue. @@ -267,7 +329,7 @@ func TestALogNotIssuedIsRefusedWithTheReason(t *testing.T) { } { _, err := logAsk(other, verb, params) if err == nil || !strings.Contains(err.Error(), `other keeps no log called "changes"`) || - !strings.Contains(err.Error(), "issued: none") { + !strings.Contains(err.Error(), "issued: issues.changes") { t.Errorf("%s of a log not issued: %v", verb, err) } } diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index c76327d..de7f3a4 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -969,9 +969,24 @@ type logAsked struct { From uint64 `json:"from"` Keys string `json:"keys"` Limit int `json:"limit"` + // Since is a read's or a watch's start as an RFC 3339 time, in place of From (novox/hq ADR 0571). + Since string `json:"since"` } -// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (ADR 0297). The runtime +// since is the asked start time, zero when none. +func (a logAsked) since() (time.Time, error) { + if a.Since == "" { + return time.Time{}, nil + } + at, err := time.Parse(time.RFC3339Nano, a.Since) + if err != nil { + return time.Time{}, fmt.Errorf("since is an RFC 3339 time (2026-10-29T09:30:00Z), not %q", a.Since) + } + return at, nil +} + +// Log answers a bundle's `append`, `read`, `newest`, `last` and `info` on its module's log and on a log its module reads +// as `.` (ADR 0297, ADR 0571). The runtime // refuses, with the reason, a log the module was not issued; an append whose `expect` is stale is // refused with the code of a lost state update. func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, error) { @@ -989,11 +1004,22 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e } answer = map[string]any{"sequence": seq} case "read": - page, err := conn.LogRead(b.module, asked.Log, asked.From, asked.Keys, asked.Limit) + since, err := asked.since() + if err != nil { + return nil, err + } + page, err := conn.LogRead(b.module, asked.Log, asked.From, since, asked.Keys, asked.Limit) if err != nil { return nil, err } answer = page + case "newest": + // The newest entry of each key the pattern names (mesh-sdk process.LogReader.Newest). + entries, err := conn.LogNewest(b.module, asked.Log, asked.Keys) + if err != nil { + return nil, err + } + answer = entries case "last": entry, err := conn.LogLast(b.module, asked.Log, asked.Key) if err != nil { @@ -1015,6 +1041,26 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e return json.Marshal(answer) } +// LogWatch hands a bundle each entry of a log, `{log, keys, from, since}`: from a sequence, from a time, or with neither +// each new one, one at a time (novox/hq ADR 0571; mesh-sdk process.LogReader.Watch). +func (b *moduleBus) LogWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { + var asked logAsked + if err := json.Unmarshal(params, &asked); err != nil || asked.Log == "" { + return nil, fmt.Errorf("mesh/log.watch names no log: {log, keys, from, since}") + } + since, err := asked.since() + if err != nil { + return nil, err + } + return b.all.conn.LogWatch(b.module, asked.Log, asked.Keys, asked.From, since, func(e bus.LogEntry) error { + raw, err := json.Marshal(e) + if err != nil { + return err + } + return deliver(raw) + }) +} + // Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the // key a pattern with `*` and `**`, empty for every key. func (b *moduleBus) Watch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { diff --git a/node-tools/module.json b/node-tools/module.json index e162449..3b52e30 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -67,6 +67,9 @@ "reads": [ "mesh-provisioner.provisions", "mesh-provisioner.plans", - "mesh-delivery-assigner.seat-holders" + "mesh-delivery-assigner.seat-holders", + "mesh-delivery-assigner.placement", + "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.manifests" ] }