A calls package of its own for objects, streams, work queues and ledgers (issue 1966 S4)

ADR 0595 §9, ADR 0597 §10, design 25 § 12.5: the request and answer types and method names of the bundle-pipe calls the tool runner serves, apply naming the tool call it serves (ADR 0597 §7) and answering taken, its sequence and state behind. It imports nothing of mesh-sdk, so a change rebuilds only its importers; the package list names it.
2026-11-03 13:23:23 +01:00
parent e9c319c5a0
commit 510d983cc6
2 changed files with 379 additions and 1 deletions
@@ -0,0 +1,342 @@
+// Package buscalls is the contract of the calls a bundle makes on its own bus state through the tool runner, over the
+// bundle's pipe (novox/hq ADR 0595 §9, ADR 0597 §10, design 25 § 12.5): the object store's, a stream's read-back and
+// fill, a work queue's size and dead letters, and the ledger's. Each call names an entry by the local name its module
+// asks for it by in `requires`; the tool runner serves it from the module's membership's bindings, its own entries
+// only. Only the shapes and the method names are here: what the tool runner refuses, and why, is its own.
+//
+// It is a package of its own, imported only by the modules that use these calls and by the tool runner, so a change
+// here rebuilds those and not the module library's importers (ADR 0595 §9). It imports nothing of mesh-sdk.
+package buscalls
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// The methods a bundle asks the tool runner, one per call.
+const (
+	ObjectsPut    = "mesh/objects.put"
+	ObjectsGet    = "mesh/objects.get"
+	ObjectsList   = "mesh/objects.list"
+	ObjectsDelete = "mesh/objects.delete"
+	ObjectsInfo   = "mesh/objects.info"
+	ObjectsWatch  = "mesh/objects.watch"
+
+	StreamRead = "mesh/stream.read"
+	StreamFill = "mesh/stream.fill"
+
+	WorkQueueSize = "mesh/workqueue.size"
+	WorkQueueDead = "mesh/workqueue.dead"
+
+	LedgerApply   = "mesh/ledger.apply"
+	LedgerGet     = "mesh/ledger.get"
+	LedgerHistory = "mesh/ledger.history"
+	LedgerWatch   = "mesh/ledger.watch"
+	LedgerQuery   = "mesh/ledger.query"
+	// The ledger's repairs, which the tool runner also serves as tools of every module owning a ledger, its own
+	// ledgers only (design 25 § 12.5).
+	LedgerRebuild  = "mesh/ledger.rebuild"
+	LedgerReindex  = "mesh/ledger.reindex"
+	LedgerSnapshot = "mesh/ledger.snapshot"
+	LedgerLag      = "mesh/ledger.lag"
+)
+
+// The methods the tool runner sends the bundle for a watch it asked, each naming the watch by its id, as `mesh/log`
+// does for a log.
+const (
+	ObjectsWatched = "mesh/objects"
+	LedgerWatched  = "mesh/ledger"
+)
+
+// Fill is how much an entry holds against its cap: its messages (or objects), its bytes and its cap in bytes.
+type Fill struct {
+	Count    uint64 `json:"count"`
+	Bytes    uint64 `json:"bytes"`
+	MaxBytes uint64 `json:"max_bytes"`
+}
+
+// Done is the answer of a call that answers nothing but that it was done.
+type Done struct{}
+
+// --- the object store
+
+// ObjectsPutRequest stores one object whole under its name, replacing one of that name.
+type ObjectsPutRequest struct {
+	Objects     string `json:"objects"`
+	Name        string `json:"name"`
+	Data        []byte `json:"data"`
+	Description string `json:"description,omitempty"`
+}
+
+// ObjectsNameRequest names one object of an object store: get, delete.
+type ObjectsNameRequest struct {
+	Objects string `json:"objects"`
+	Name    string `json:"name"`
+}
+
+// ObjectsRequest names an object store: list, info.
+type ObjectsRequest struct {
+	Objects string `json:"objects"`
+}
+
+// ObjectsWatchRequest asks the tool runner to send every object stored or deleted, as ObjectsChange under the watch's id.
+type ObjectsWatchRequest struct {
+	Objects string `json:"objects"`
+	Watch   string `json:"watch"`
+}
+
+// ObjectInfo is one object as its store holds it: put's answer, and each of list's.
+type ObjectInfo struct {
+	Name        string    `json:"name"`
+	Size        uint64    `json:"size"`
+	Chunks      uint32    `json:"chunks"`
+	Digest      string    `json:"digest"`
+	Description string    `json:"description,omitempty"`
+	Modified    time.Time `json:"modified"`
+	Deleted     bool      `json:"deleted,omitempty"`
+}
+
+// ObjectData is get's answer: the object's info and its bytes.
+type ObjectData struct {
+	Info ObjectInfo `json:"info"`
+	Data []byte     `json:"data"`
+}
+
+// ObjectsListed is list's answer: every object the store holds.
+type ObjectsListed struct {
+	Objects []ObjectInfo `json:"objects"`
+}
+
+// ObjectsChange is one object stored or deleted, sent under ObjectsWatched; Current says it was there when the watch
+// began.
+type ObjectsChange struct {
+	Watch   string     `json:"watch"`
+	Object  ObjectInfo `json:"object"`
+	Current bool       `json:"current"`
+}
+
+// --- a stream
+
+// StreamReadRequest reads a module's own stream from a sequence on, at most Limit messages, Subjects narrowing it.
+type StreamReadRequest struct {
+	Stream   string `json:"stream"`
+	From     uint64 `json:"from"`
+	Subjects string `json:"subjects,omitempty"`
+	Limit    int    `json:"limit,omitempty"`
+}
+
+// StreamRequest names a stream: fill.
+type StreamRequest struct {
+	Stream string `json:"stream"`
+}
+
+// StreamMessage is one message of a stream.
+type StreamMessage struct {
+	Sequence uint64          `json:"sequence"`
+	Subject  string          `json:"subject"`
+	Body     json.RawMessage `json:"body"`
+	At       time.Time       `json:"at"`
+}
+
+// StreamPage is read's answer: the messages in order, the sequence to read on from (0 when nothing follows) and the
+// stream's last sequence when it was read.
+type StreamPage struct {
+	Messages []StreamMessage `json:"messages"`
+	Next     uint64          `json:"next"`
+	Last     uint64          `json:"last"`
+}
+
+// StreamFilled is fill's answer: the stream's fill and its first and last sequence.
+type StreamFilled struct {
+	Fill
+	First uint64 `json:"first"`
+	Last  uint64 `json:"last"`
+}
+
+// --- a work queue
+
+// WorkQueueRequest names a work queue: size.
+type WorkQueueRequest struct {
+	Queue string `json:"queue"`
+}
+
+// WorkQueueDeadRequest asks a work queue's dead letters, at most Limit of them.
+type WorkQueueDeadRequest struct {
+	Queue string `json:"queue"`
+	Limit int    `json:"limit,omitempty"`
+}
+
+// WorkQueueSized is size's answer: the messages waiting to be taken, against the queue's cap.
+type WorkQueueSized struct {
+	Fill
+}
+
+// DeadLetter is one message delivered as often as its queue allows and never settled.
+type DeadLetter struct {
+	Sequence   uint64          `json:"sequence"`
+	Subject    string          `json:"subject"`
+	Deliveries int             `json:"deliveries"`
+	Body       json.RawMessage `json:"body,omitempty"`
+	At         time.Time       `json:"at"`
+}
+
+// WorkQueueDeadLetters is dead's answer.
+type WorkQueueDeadLetters struct {
+	Dead []DeadLetter `json:"dead"`
+}
+
+// --- the ledger (ADR 0597)
+
+// ApplyRequest applies one operation to one instance of a ledger. Call is the id of the tool call the operation
+// serves, whose caller the tool runner proved and stamps as the author (ADR 0597 §7); empty, the operation carries the
+// owning module's account alone. An author the bundle writes into Body is never taken as the author.
+type ApplyRequest struct {
+	Ledger    string          `json:"ledger"`
+	Instance  string          `json:"instance"`
+	Operation string          `json:"operation"`
+	Body      json.RawMessage `json:"body,omitempty"`
+	Call      string          `json:"call,omitempty"`
+}
+
+// Applied is apply's answer: the operation was taken, with its sequence in the instance's history; StateBehind says
+// the state could not follow because its bucket is full, and catches up at the next apply after room is made (ADR
+// 0597 §6).
+type Applied struct {
+	Taken       bool   `json:"taken"`
+	Sequence    uint64 `json:"sequence"`
+	StateBehind bool   `json:"state_behind,omitempty"`
+}
+
+// Author is who an operation is by, as `apply` stamped it (ADR 0597 §7): the owning module's account, and the caller
+// of the tool call it served, as the tool runner proved it.
+type Author struct {
+	Account string `json:"account"`
+	Caller  string `json:"caller,omitempty"`
+}
+
+// InstanceRequest names one instance of a ledger: get, snapshot.
+type InstanceRequest struct {
+	Ledger   string `json:"ledger"`
+	Instance string `json:"instance"`
+}
+
+// InstanceState is get's answer: the instance's state and the sequence of the last operation it holds.
+type InstanceState struct {
+	Instance string          `json:"instance"`
+	State    json.RawMessage `json:"state"`
+	Sequence uint64          `json:"sequence"`
+}
+
+// HistoryRequest reads an instance's operations from a sequence on, at most Limit of them.
+type HistoryRequest struct {
+	Ledger   string `json:"ledger"`
+	Instance string `json:"instance"`
+	From     uint64 `json:"from,omitempty"`
+	Limit    int    `json:"limit,omitempty"`
+}
+
+// Operation is one operation as the history keeps it.
+type Operation struct {
+	Sequence  uint64          `json:"sequence"`
+	Instance  string          `json:"instance"`
+	Operation string          `json:"operation"`
+	Body      json.RawMessage `json:"body,omitempty"`
+	Author    Author          `json:"author"`
+	At        time.Time       `json:"at"`
+}
+
+// HistoryPage is history's answer: the operations oldest first, the sequence to read on from (0 when nothing
+// follows) and the instance's last sequence when it was read.
+type HistoryPage struct {
+	Operations []Operation `json:"operations"`
+	Next       uint64      `json:"next"`
+	Last       uint64      `json:"last"`
+}
+
+// LedgerWatchRequest asks the tool runner to send every change of the instances Instances names (an instance, with
+// `*` for one name and `>` for the rest) as InstanceChange under the watch's id: the current states first.
+type LedgerWatchRequest struct {
+	Ledger    string `json:"ledger"`
+	Instances string `json:"instances"`
+	Watch     string `json:"watch"`
+}
+
+// InstanceChange is one instance's new state, sent under LedgerWatched; Current says it was the state when the watch
+// began.
+type InstanceChange struct {
+	Watch    string          `json:"watch"`
+	Instance string          `json:"instance"`
+	State    json.RawMessage `json:"state"`
+	Sequence uint64          `json:"sequence"`
+	Current  bool            `json:"current"`
+}
+
+// QueryRequest reads a declared index of a ledger: the entries whose key Key names (a key, with `*` for one name and
+// `>` for the rest; empty is every key), at most Limit of them.
+type QueryRequest struct {
+	Ledger string `json:"ledger"`
+	Index  string `json:"index"`
+	Key    string `json:"key,omitempty"`
+	Limit  int    `json:"limit,omitempty"`
+}
+
+// IndexEntry is one entry of an index.
+type IndexEntry struct {
+	Key   string          `json:"key"`
+	Value json.RawMessage `json:"value"`
+}
+
+// QueryAnswer is query's answer.
+type QueryAnswer struct {
+	Entries []IndexEntry `json:"entries"`
+}
+
+// RebuildFrom is what a rebuild derives the state from: the history, or the snapshot and the operations after it.
+const (
+	RebuildFromHistory  = "history"
+	RebuildFromSnapshot = "snapshot"
+)
+
+// RebuildRequest derives an instance's state again (every instance's when Instance is empty), From the history or the
+// snapshot.
+type RebuildRequest struct {
+	Ledger   string `json:"ledger"`
+	Instance string `json:"instance,omitempty"`
+	From     string `json:"from,omitempty"`
+}
+
+// ReindexRequest builds a declared index again from the history.
+type ReindexRequest struct {
+	Ledger string `json:"ledger"`
+	Index  string `json:"index"`
+}
+
+// LedgerRequest names a ledger: lag.
+type LedgerRequest struct {
+	Ledger string `json:"ledger"`
+}
+
+// Repaired is the answer of rebuild, reindex and snapshot: how many instances (or index entries) it wrote.
+type Repaired struct {
+	Written int `json:"written"`
+}
+
+// InstanceLag is one instance whose state or snapshot is behind its history: the sequences each holds.
+type InstanceLag struct {
+	Instance string `json:"instance"`
+	History  uint64 `json:"history"`
+	State    uint64 `json:"state"`
+	Snapshot uint64 `json:"snapshot"`
+}
+
+// IndexLag is how far one declared index is behind the history.
+type IndexLag struct {
+	Index  string `json:"index"`
+	Behind uint64 `json:"behind"`
+}
+
+// Lagged is lag's answer: every instance and index behind its history; none when nothing is behind.
+type Lagged struct {
+	Instances []InstanceLag `json:"instances,omitempty"`
+	Indexes   []IndexLag    `json:"indexes,omitempty"`
+}
@@ -1,14 +1,19 @@
 package stdio

 import (
+	"encoding/json"
 	"go/ast"
 	"go/parser"
 	"go/token"
 	"io/fs"
 	"path/filepath"
+	"reflect"
 	"slices"
+	"strconv"
 	"strings"
 	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 )

 // meshSDKPackages is every package mesh-sdk holds (novox/hq ADR 0593 §1): what every module shares, the value shapes
@@ -16,7 +21,7 @@ import (
 // Logic only some modules need is its owner module's package. A package added here is a decision to say in the pull
 // request: does every module share it, or is it a shape more than one part reads?
 var meshSDKPackages = []string{
-	".", "announce", "announcetest", "asker", "asks", "buildstep", "buscompose", "busretry", "calladdress", "catalogue",
+	".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busretry", "calladdress", "catalogue",
 	"composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership",
 	"objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions",
 	"sdkversion", "seal", "testbus",
@@ -56,6 +61,16 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
 		if err != nil {
 			return err
 		}
+		// 1966/AC5: the calls package imports nothing of mesh-sdk but the catalogue's types, so a change here rebuilds
+		// only its importers (ADR 0595 §9).
+		if dir == "buscalls" {
+			for _, imp := range file.Imports {
+				p, _ := strconv.Unquote(imp.Path.Value)
+				if strings.HasPrefix(p, sdkModule) && p != sdkModule+"/catalogue" {
+					t.Errorf("%s imports %s: the calls package imports nothing of mesh-sdk but catalogue (ADR 0595 §9)", path, p)
+				}
+			}
+		}
 		for _, decl := range file.Decls {
 			fn, ok := decl.(*ast.FuncDecl)
 			if !ok {
@@ -90,4 +105,25 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
 			t.Errorf("mesh-sdk's list keeps %s, which is gone: take it off the list", want)
 		}
 	}
+
+	// 1966/AC5: an apply naming the tool call it serves, and its answer with the state behind, round-trip.
+	for _, c := range []struct {
+		name     string
+		in, back any
+	}{
+		{"apply", &buscalls.ApplyRequest{Ledger: "records", Instance: "decision/0290", Operation: "submit",
+			Body: json.RawMessage(`{"why":"x"}`), Call: "call-17"}, &buscalls.ApplyRequest{}},
+		{"applied", &buscalls.Applied{Taken: true, Sequence: 5, StateBehind: true}, &buscalls.Applied{}},
+	} {
+		raw, err := json.Marshal(c.in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := json.Unmarshal(raw, c.back); err != nil || !reflect.DeepEqual(c.in, c.back) {
+			t.Errorf("%s read back as %+v from %s (%v)", c.name, c.back, raw, err)
+		}
+	}
 }
+
+// sdkModule is mesh-sdk's Go module path.
+const sdkModule = "git.novox.be/novox/mesh-sdk/go"