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" ] }