From d6c516ab445a84ead70dccb850baeb30da588c73 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:34:32 +0100 Subject: [PATCH] Serve a bundle its own bound objects, streams and work queues on its pipe (issue 1966 T1) --- node-tools/go.mod | 2 + node-tools/go.sum | 2 - node-tools/internal/bus/bindings.go | 398 ++++++++++++++++++ node-tools/internal/bus/bindings_test.go | 112 +++++ node-tools/internal/bus/bus.go | 3 + .../internal/launch/event_answer_test.go | 4 + node-tools/internal/launch/launch.go | 45 ++ node-tools/internal/runtime/busstate.go | 102 +++++ 8 files changed, 666 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/bus/bindings.go create mode 100644 node-tools/internal/bus/bindings_test.go create mode 100644 node-tools/internal/runtime/busstate.go diff --git a/node-tools/go.mod b/node-tools/go.mod index 4b2d0b9..f2cc3df 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -20,3 +20,5 @@ require ( golang.org/x/crypto v0.57.0 // indirect golang.org/x/time v0.15.0 // indirect ) + +replace git.novox.be/novox/mesh-sdk/go => ../../mesh-sdk/go diff --git a/node-tools/go.sum b/node-tools/go.sum index 43cc4ec..98dc72c 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,5 +1,3 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 h1:aI1DhZAuqVBXbybSqhLNOGX5rUOrn9T0cHNCDUQF7IQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128/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/bindings.go b/node-tools/internal/bus/bindings.go new file mode 100644 index 0000000..8937a61 --- /dev/null +++ b/node-tools/internal/bus/bindings.go @@ -0,0 +1,398 @@ +package bus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats.go/jetstream" +) + +// A module's bus state in the new form (novox/hq ADR 0595, design 25 § 12.5; issue 1966 T1): an object store, a stream, +// a work queue, a ledger, each asked for in `requires` by a local name and bound in the module's membership by the bus +// names the bus module created. **A bundle reaches only its own module's bindings**: one account per machine carries +// every module on it, so what a module may reach is decided here, from its membership, and refused with the reason +// before anything is sent, as its state is (state.go). + +// Binding is the module's binding of a local name, of the storage type storage, or the reason there is none. +func (c *Conn) Binding(module, local, storage string) (membership.BusBinding, error) { + m := c.Membership(module) + if m == nil { + return membership.BusBinding{}, fmt.Errorf("%s has no membership issued on %s yet, so none of its bus state is "+ + "reachable until the mesh issues one (novox/hq ADR 0595)", module, c.node) + } + var names []string + for _, b := range m.Bindings { + if b.Local != local { + names = append(names, b.Local+" ("+b.Storage+")") + continue + } + if b.Storage != storage { + return membership.BusBinding{}, fmt.Errorf("%s's %q is bound with the storage type %s, not %s (novox/hq ADR 0595)", + module, local, b.Storage, storage) + } + return b, nil + } + sort.Strings(names) + bound := "none" + if len(names) > 0 { + bound = strings.Join(names, ", ") + } + return membership.BusBinding{}, fmt.Errorf("%s was bound no %s called %q: a module reaches only the bus state its "+ + "manifest asks for in `requires`, by its local name, once the mesh created it; it was bound: %s (novox/hq ADR 0595)", + module, storage, local, bound) +} + +// Bindings is every binding a module's membership carries now. +func (c *Conn) Bindings(module string) []membership.BusBinding { + if m := c.Membership(module); m != nil { + return append([]membership.BusBinding(nil), m.Bindings...) + } + return nil +} + +// jetStream is a JetStream handle on this connection. +func (c *Conn) jetStream() (jetstream.JetStream, error) { return jetstream.New(c.nc) } + +func (c *Conn) objectStore(ctx context.Context, module, local string) (jetstream.ObjectStore, membership.BusBinding, error) { + b, err := c.Binding(module, local, catalogue.StorageObjectStore) + if err != nil { + return nil, b, err + } + js, err := c.jetStream() + if err != nil { + return nil, b, err + } + obs, err := js.ObjectStore(ctx, b.ObjectStore) + if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { + return nil, b, fmt.Errorf("%s's objects %q are bound and their store %s is not on the bus", module, local, b.ObjectStore) + } + return obs, b, err +} + +func objectInfoOf(i *jetstream.ObjectInfo) buscalls.ObjectInfo { + return buscalls.ObjectInfo{Name: i.Name, Size: i.Size, Chunks: i.Chunks, Digest: i.Digest, Description: i.Description, + Modified: i.ModTime, Deleted: i.Deleted} +} + +// ObjectsPut stores one object whole under its name in the module's object store. +func (c *Conn) ObjectsPut(module string, r buscalls.ObjectsPutRequest) (buscalls.ObjectInfo, error) { + if r.Name == "" { + return buscalls.ObjectInfo{}, errors.New("an object is put under a name") + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return buscalls.ObjectInfo{}, err + } + info, err := obs.PutBytes(ctx, r.Name, r.Data) + if err != nil { + return buscalls.ObjectInfo{}, err + } + if r.Description != "" { + if err := obs.UpdateMeta(ctx, r.Name, jetstream.ObjectMeta{Name: r.Name, Description: r.Description}); err != nil { + return buscalls.ObjectInfo{}, err + } + if info, err = obs.GetInfo(ctx, r.Name); err != nil { + return buscalls.ObjectInfo{}, err + } + } + return objectInfoOf(info), nil +} + +// ObjectsGet is one object of the module's object store, its info and its bytes; nil when there is none. +func (c *Conn) ObjectsGet(module string, r buscalls.ObjectsNameRequest) (*buscalls.ObjectData, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return nil, err + } + res, err := obs.Get(ctx, r.Name) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + defer res.Close() + info, err := res.Info() + if err != nil { + return nil, err + } + data, err := obs.GetBytes(ctx, r.Name) + if err != nil { + return nil, err + } + return &buscalls.ObjectData{Info: objectInfoOf(info), Data: data}, nil +} + +// ObjectsDelete deletes one object of the module's object store; one that is not there is not an error. +func (c *Conn) ObjectsDelete(module string, r buscalls.ObjectsNameRequest) error { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return err + } + if err := obs.Delete(ctx, r.Name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) { + return err + } + return nil +} + +// ObjectsList is every object the module's object store holds. +func (c *Conn) ObjectsList(module string, r buscalls.ObjectsRequest) (buscalls.ObjectsListed, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return buscalls.ObjectsListed{}, err + } + out := buscalls.ObjectsListed{Objects: []buscalls.ObjectInfo{}} + infos, err := obs.List(ctx) + if errors.Is(err, jetstream.ErrNoObjectsFound) { + return out, nil + } + if err != nil { + return out, err + } + for _, i := range infos { + out.Objects = append(out.Objects, objectInfoOf(i)) + } + return out, nil +} + +// ObjectsInfo is how full the module's object store is: its objects, its bytes and its cap. +func (c *Conn) ObjectsInfo(module string, r buscalls.ObjectsRequest) (buscalls.Fill, error) { + listed, err := c.ObjectsList(module, r) + if err != nil { + return buscalls.Fill{}, err + } + b, _ := c.Binding(module, r.Objects, catalogue.StorageObjectStore) + fill, err := c.streamFill("OBJ_" + b.ObjectStore) + if err != nil { + return buscalls.Fill{}, err + } + fill.Count = uint64(len(listed.Objects)) + return fill.Fill, nil +} + +// ObjectsWatch hands deliver every object of the module's object store as it is, then each stored or deleted. +func (c *Conn) ObjectsWatch(module string, r buscalls.ObjectsWatchRequest, deliver func(buscalls.ObjectsChange) error) (func(), error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + obs, _, err := c.objectStore(ctx, module, r.Objects) + cancel() + if err != nil { + return nil, err + } + watchCtx, stop := context.WithCancel(context.Background()) + w, err := obs.Watch(watchCtx, jetstream.IncludeHistory()) + if err != nil { + stop() + return nil, err + } + current := true + for current { + select { + case i, ok := <-w.Updates(): + if !ok { + stop() + return nil, errors.New("the object store's watch ended before its current objects were handed over") + } + if i == nil { + current = false + continue + } + if err := deliver(buscalls.ObjectsChange{Watch: r.Watch, Object: objectInfoOf(i), Current: true}); err != nil { + c.Logf("[mesh-tools] %s did not take an object of %s: %v", module, r.Objects, err) + } + case <-time.After(StateTimeout): + _ = w.Stop() + stop() + return nil, fmt.Errorf("the object store %s did not hand its current objects in %s", r.Objects, StateTimeout) + } + } + go func() { + for i := range w.Updates() { + if i == nil { + continue + } + if err := deliver(buscalls.ObjectsChange{Watch: r.Watch, Object: objectInfoOf(i)}); err != nil { + c.Logf("[mesh-tools] %s did not take an object of %s: %v", module, r.Objects, err) + } + } + }() + return func() { _ = w.Stop(); stop() }, nil +} + +// streamFill is a stream's fill and its first and last sequence, by its name on the bus. +func (c *Conn) streamFill(name string) (buscalls.StreamFilled, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := c.jetStream() + if err != nil { + return buscalls.StreamFilled{}, err + } + s, err := js.Stream(ctx, name) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return buscalls.StreamFilled{}, fmt.Errorf("the stream %s is bound and not on the bus", name) + } + if err != nil { + return buscalls.StreamFilled{}, err + } + info, err := s.Info(ctx) + if err != nil { + return buscalls.StreamFilled{}, err + } + max := uint64(0) + if info.Config.MaxBytes > 0 { + max = uint64(info.Config.MaxBytes) + } + return buscalls.StreamFilled{Fill: buscalls.Fill{Count: info.State.Msgs, Bytes: info.State.Bytes, MaxBytes: max}, + First: info.State.FirstSeq, Last: info.State.LastSeq}, nil +} + +// streamBinding is the module's binding of a stream by its local name, a work queue among them. +func (c *Conn) streamBinding(module, local string) (membership.BusBinding, error) { + return c.Binding(module, local, catalogue.StorageStream) +} + +// StreamFill is how full the module's stream is. +func (c *Conn) StreamFill(module string, r buscalls.StreamRequest) (buscalls.StreamFilled, error) { + b, err := c.streamBinding(module, r.Stream) + if err != nil { + return buscalls.StreamFilled{}, err + } + return c.streamFill(b.Stream) +} + +// StreamReadMost is the most messages one read answers. +const StreamReadMost = 1000 + +// StreamRead is the module's stream's messages from a sequence on, Subjects narrowing them, at most Limit of them. +func (c *Conn) StreamRead(module string, r buscalls.StreamReadRequest) (buscalls.StreamPage, error) { + b, err := c.streamBinding(module, r.Stream) + if err != nil { + return buscalls.StreamPage{}, err + } + return c.readStream(b.Stream, r.Subjects, r.From, r.Limit) +} + +// readStream reads a stream by its bus name from a sequence on, the messages on the subjects filter names (every +// subject when empty), at most limit of them. +func (c *Conn) readStream(name, filter string, from uint64, limit int) (buscalls.StreamPage, error) { + if limit <= 0 || limit > StreamReadMost { + limit = StreamReadMost + } + if filter == "" { + filter = ">" + } + if from == 0 { + from = 1 + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := c.jetStream() + if err != nil { + return buscalls.StreamPage{}, err + } + s, err := js.Stream(ctx, name) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return buscalls.StreamPage{}, fmt.Errorf("the stream %s is bound and not on the bus", name) + } + if err != nil { + return buscalls.StreamPage{}, err + } + info, err := s.Info(ctx) + if err != nil { + return buscalls.StreamPage{}, err + } + page := buscalls.StreamPage{Messages: []buscalls.StreamMessage{}, Last: info.State.LastSeq} + for seq := from; seq <= info.State.LastSeq && len(page.Messages) < limit; { + msg, err := s.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(filter)) + if errors.Is(err, jetstream.ErrMsgNotFound) { + break + } + if err != nil { + return page, err + } + body := json.RawMessage(msg.Data) + if !json.Valid(msg.Data) { + body, _ = json.Marshal(string(msg.Data)) + } + page.Messages = append(page.Messages, buscalls.StreamMessage{Sequence: msg.Sequence, Subject: msg.Subject, + Body: body, At: msg.Time}) + seq = msg.Sequence + 1 + if len(page.Messages) == limit && seq <= info.State.LastSeq { + page.Next = seq + } + } + return page, nil +} + +// WorkQueueSize is how many messages wait in the module's work queue, against its cap. +func (c *Conn) WorkQueueSize(module string, r buscalls.WorkQueueRequest) (buscalls.WorkQueueSized, error) { + b, err := c.streamBinding(module, r.Queue) + if err != nil { + return buscalls.WorkQueueSized{}, err + } + fill, err := c.streamFill(b.Stream) + if err != nil { + return buscalls.WorkQueueSized{}, err + } + return buscalls.WorkQueueSized{Fill: fill.Fill}, nil +} + +// WorkQueueDead is the module's work queue's dead letters: in a work queue a message is gone once settled, so one still +// kept below a consumer's acknowledgement floor was delivered as often as the consumer allows and never settled. +func (c *Conn) WorkQueueDead(module string, r buscalls.WorkQueueDeadRequest) (buscalls.WorkQueueDeadLetters, error) { + b, err := c.streamBinding(module, r.Queue) + if err != nil { + return buscalls.WorkQueueDeadLetters{}, err + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := c.jetStream() + if err != nil { + return buscalls.WorkQueueDeadLetters{}, err + } + s, err := js.Stream(ctx, b.Stream) + if err != nil { + return buscalls.WorkQueueDeadLetters{}, fmt.Errorf("the work queue %s: %w", b.Stream, err) + } + out := buscalls.WorkQueueDeadLetters{Dead: []buscalls.DeadLetter{}} + consumers := s.ListConsumers(ctx) + for ci := range consumers.Info() { + floor := ci.AckFloor.Stream + if ci.Config.MaxDeliver <= 0 || floor == 0 { + continue + } + page, err := c.readStream(b.Stream, ci.Config.FilterSubject, 1, r.Limit) + if err != nil { + return out, err + } + for _, m := range page.Messages { + if m.Sequence > floor { + break + } + out.Dead = append(out.Dead, buscalls.DeadLetter{Sequence: m.Sequence, Subject: m.Subject, + Deliveries: ci.Config.MaxDeliver, Body: m.Body, At: m.At}) + } + } + if err := consumers.Err(); err != nil { + return out, err + } + if r.Limit > 0 && len(out.Dead) > r.Limit { + out.Dead = out.Dead[:r.Limit] + } + return out, nil +} diff --git a/node-tools/internal/bus/bindings_test.go b/node-tools/internal/bus/bindings_test.go new file mode 100644 index 0000000..81c25a0 --- /dev/null +++ b/node-tools/internal/bus/bindings_test.go @@ -0,0 +1,112 @@ +package bus_test + +import ( + "context" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// 1966/AC12: given a bundle whose membership binds objects `files`, when it puts and gets `a`, then it reads back what it +// put; a local name it was not bound is refused with the reason, and so is another module's binding and a binding of +// another storage type. A stream and a work queue bound alike answer their read-back, fill and size. +func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) { + mesh := mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx := context.Background() + if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_files"}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "m_events", Subjects: []string{"mesh.state.m.events.>"}}); err != nil { + t.Fatal(err) + } + for _, s := range []string{"mesh.state.m.events.one", "mesh.state.m.events.two"} { + if _, err := js.Publish(ctx, s, []byte(`{"n":1}`)); err != nil { + t.Fatal(err) + } + } + m := mt.MembershipOf("m", "anchor", false, nil) + m.Bindings = []membership.BusBinding{ + {Local: "files", Storage: catalogue.StorageObjectStore, Template: catalogue.TemplateObjects, ObjectStore: "m_files"}, + {Local: "events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, Stream: "m_events", + Subject: "mesh.state.m.events"}, + } + other := mt.MembershipOf("other", "anchor", false, nil) + mesh.Issue(t, m) + mesh.Issue(t, other) + conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + conn.Follow("m") + conn.Follow("other") + mt.Until(t, func() error { + if conn.Membership("m") == nil || conn.Membership("other") == nil { + return errors.New("not issued yet") + } + return nil + }) + + if _, err := conn.ObjectsPut("m", buscalls.ObjectsPutRequest{Objects: "files", Name: "a", Data: []byte("what was put")}); err != nil { + t.Fatal(err) + } + got, err := conn.ObjectsGet("m", buscalls.ObjectsNameRequest{Objects: "files", Name: "a"}) + if err != nil || got == nil || string(got.Data) != "what was put" || got.Info.Name != "a" { + t.Fatalf("read back %+v, %v", got, err) + } + if listed, err := conn.ObjectsList("m", buscalls.ObjectsRequest{Objects: "files"}); err != nil || len(listed.Objects) != 1 { + t.Fatalf("listed %+v, %v", listed, err) + } + page, err := conn.StreamRead("m", buscalls.StreamReadRequest{Stream: "events", Limit: 1}) + if err != nil || len(page.Messages) != 1 || page.Next != 2 || page.Last != 2 { + t.Fatalf("read %+v, %v", page, err) + } + if fill, err := conn.StreamFill("m", buscalls.StreamRequest{Stream: "events"}); err != nil || fill.Count != 2 { + t.Fatalf("fill %+v, %v", fill, err) + } + + for _, c := range []struct { + name, module, refusal string + call func(module string) error + }{ + {"a local name not bound", "m", `bound no object-store called "photos"`, func(module string) error { + _, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "photos", Name: "a"}) + return err + }}, + {"another module's binding", "other", `other was bound no object-store called "files"`, func(module string) error { + _, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "files", Name: "a"}) + return err + }}, + {"a binding of another storage type", "m", `"events" is bound with the storage type stream, not object-store`, func(module string) error { + _, err := conn.ObjectsList(module, buscalls.ObjectsRequest{Objects: "events"}) + return err + }}, + {"a work queue not bound", "m", `bound no stream called "jobs"`, func(module string) error { + _, err := conn.WorkQueueSize(module, buscalls.WorkQueueRequest{Queue: "jobs"}) + return err + }}, + } { + t.Run(c.name, func(t *testing.T) { + err := c.call(c.module) + if err == nil || !strings.Contains(err.Error(), c.refusal) { + t.Fatalf("refused with %v, want the reason %q", err, c.refusal) + } + }) + } +} diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index fa1a7c3..8027e53 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -78,6 +78,9 @@ type Membership struct { // Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2): // the tool runner's alone, composed from its manifest with the grant that lets it subscribe them. Waits []WaitIssued `json:"waits,omitempty"` + // Bindings are the module's bus state in the new form, by local name (novox/hq ADR 0595 §1, mesh-sdk + // membership.BusBinding): its object stores, streams, work queues and ledgers, its own alone (bindings.go). + Bindings []membership.BusBinding `json:"bindings,omitempty"` } // Served is an address a tool is answered on; `{tool}` stands for the tool's name. diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 7ccfc47..6746b36 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -48,6 +48,10 @@ func (b *subscribing) Provision(string, json.RawMessage) (json.RawMessage, error func (b *subscribing) ProvisionWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { return func() {}, nil } +func (b *subscribing) BusState(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } +func (b *subscribing) BusStateWatch(string, json.RawMessage, func(json.RawMessage) error) (func(), error) { + return func() {}, nil +} func (b *subscribing) Subscribe(d func(json.RawMessage) error) error { b.mu.Lock() defer b.mu.Unlock() diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 43f3fff..0b91f71 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -19,6 +19,7 @@ import ( "syscall" "time" + "git.novox.be/novox/mesh-sdk/go/buscalls" "golang.org/x/sys/unix" "github.com/novox/mesh-tools/node-tools/internal/wire" @@ -91,8 +92,26 @@ type Bus interface { // (novox/hq ADR 0349 §14, ADR 0370 §4): its config, and its bindings' values that are not secret. Provision(verb string, params json.RawMessage) (json.RawMessage, error) ProvisionWatch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) + // BusState answers a bundle's calls on its module's bus state in the new form (mesh-sdk buscalls; novox/hq ADR 0595 + // §9, design 25 § 12.5): its objects, a stream's read-back and fill, a work queue's size and dead letters, and its + // ledgers, by the method the bundle asked, from its own module's bindings alone. + BusState(method string, params json.RawMessage) (json.RawMessage, error) + // BusStateWatch hands deliver each change of an objects or a ledger watch, the current ones first. + BusStateWatch(method string, params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) } +// busStateCalls are the methods of mesh-sdk buscalls a bundle asks and BusState answers. +var busStateCalls = map[string]bool{ + buscalls.ObjectsPut: true, buscalls.ObjectsGet: true, buscalls.ObjectsList: true, buscalls.ObjectsDelete: true, + buscalls.ObjectsInfo: true, buscalls.StreamRead: true, buscalls.StreamFill: true, buscalls.WorkQueueSize: true, + buscalls.WorkQueueDead: true, buscalls.LedgerApply: true, buscalls.LedgerGet: true, buscalls.LedgerHistory: true, + buscalls.LedgerQuery: true, buscalls.LedgerRebuild: true, buscalls.LedgerReindex: true, buscalls.LedgerSnapshot: true, + buscalls.LedgerLag: true, +} + +// busStateWatches are the watches of mesh-sdk buscalls, each with the method its changes are handed to the bundle by. +var busStateWatches = map[string]string{buscalls.ObjectsWatch: buscalls.ObjectsWatched, buscalls.LedgerWatch: buscalls.LedgerWatched} + // EventTimeout bounds how long a bundle has to handle one event before it is offered again. var EventTimeout = 2 * time.Minute @@ -399,7 +418,33 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an c.mu.Unlock() } } + case buscalls.ObjectsWatch, buscalls.LedgerWatch: + // As a state's watch: each change asked of this child in order, by the method of its + // family; the change names the watch by the id the bundle gave in its request. + var stop func() + told := busStateWatches[m.Method] + stop, err = mesh.BusStateWatch(m.Method, m.Params, func(change json.RawMessage) error { + _, err := c.ask(module, told, change, 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() + } + } default: + if busStateCalls[m.Method] { + var answered json.RawMessage + if answered, err = mesh.BusState(m.Method, m.Params); err == nil { + result = answered + } + break + } if len(m.ID) > 0 { _ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID, "error": map[string]any{"code": -32601, "message": "the runtime answers no " + m.Method + " from a bundle"}}) diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go new file mode 100644 index 0000000..905d3f0 --- /dev/null +++ b/node-tools/internal/runtime/busstate.go @@ -0,0 +1,102 @@ +package runtime + +import ( + "encoding/json" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// BusState answers a bundle's calls on its module's bus state in the new form (mesh-sdk buscalls; novox/hq ADR 0595 +// §9, design 25 § 12.5; issue 1966): each as the module whose bundle asks, never a module the request names, so a +// bundle reaches its own module's bindings alone. +func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMessage, error) { + conn := b.all.conn + var answer any + var err error + switch method { + case buscalls.ObjectsPut: + var r buscalls.ObjectsPutRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.ObjectsPut(b.module, r) + } + case buscalls.ObjectsGet: + var r buscalls.ObjectsNameRequest + if err = decode(method, params, &r); err == nil { + var got *buscalls.ObjectData + if got, err = conn.ObjectsGet(b.module, r); err == nil && got == nil { + return json.RawMessage("null"), nil + } + answer = got + } + case buscalls.ObjectsDelete: + var r buscalls.ObjectsNameRequest + if err = decode(method, params, &r); err == nil { + err = conn.ObjectsDelete(b.module, r) + answer = buscalls.Done{} + } + case buscalls.ObjectsList: + var r buscalls.ObjectsRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.ObjectsList(b.module, r) + } + case buscalls.ObjectsInfo: + var r buscalls.ObjectsRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.ObjectsInfo(b.module, r) + } + case buscalls.StreamRead: + var r buscalls.StreamReadRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.StreamRead(b.module, r) + } + case buscalls.StreamFill: + var r buscalls.StreamRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.StreamFill(b.module, r) + } + case buscalls.WorkQueueSize: + var r buscalls.WorkQueueRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.WorkQueueSize(b.module, r) + } + case buscalls.WorkQueueDead: + var r buscalls.WorkQueueDeadRequest + if err = decode(method, params, &r); err == nil { + answer, err = conn.WorkQueueDead(b.module, r) + } + default: + return nil, fmt.Errorf("the runtime answers no %s", method) + } + if err != nil { + return nil, err + } + return json.Marshal(answer) +} + +// BusStateWatch hands a bundle each change of its objects or its ledger, the current ones first. +func (b *moduleBus) BusStateWatch(method string, params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { + switch method { + case buscalls.ObjectsWatch: + var r buscalls.ObjectsWatchRequest + if err := decode(method, params, &r); err != nil { + return nil, err + } + return b.all.conn.ObjectsWatch(b.module, r, func(change buscalls.ObjectsChange) error { + raw, err := json.Marshal(change) + if err != nil { + return err + } + return deliver(raw) + }) + } + return nil, fmt.Errorf("the runtime answers no %s", method) +} + +// decode reads a bus-state call's request. +func decode(method string, params json.RawMessage, into any) error { + if err := json.Unmarshal(params, into); err != nil { + return fmt.Errorf("%s's request is not of its shape (mesh-sdk buscalls): %w", method, err) + } + return nil +}