Serve a bundle its own bound objects, streams and work queues on its pipe (issue 1966 T1)

2026-11-03 13:34:32 +01:00
parent b8e42cf54c
commit d6c516ab44
8 changed files with 666 additions and 2 deletions
@@ -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
+}