R-1 rest, R-12, R-13: the capability words are mesh-sdk's, the ledger's fill and removal are answered, the artifact states the words

sent-get is under sent-objects-calls; mesh/ledger.fill answers the history's fill against its cap and mesh/ledger.remove
removes an instance's operations, snapshot, state and index entries, both under ledger-upkeep-calls; the runtime artifact's
`states` is exactly the words the runner states.

Builds: design 47 G §462, §488, §545, §587, §622, §155

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 05:02:04 +01:00
parent 64015b5952
commit 5e373af70d
10 changed files with 364 additions and 19 deletions
+76
View File
@@ -10,6 +10,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/membership"
"github.com/nats-io/nats.go"
@@ -143,6 +144,76 @@ func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, e
}
}
// Purge removes every operation of one instance from the history stream, on the module's ledger connection, the one
// granted the purge of its own ledger's streams.
func (h *busHistory) Purge(key string) error {
return purgeSubject(h.read, h.write, h.stream, h.prefix+"."+key)
}
// Fill is the history stream's fill against its cap, the cap the manifest gives the ledger.
func (h *busHistory) Fill() (buscalls.Fill, error) {
name := h.stream
if name == "" {
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
defer cancel()
js, err := h.read.JetStream()
if err != nil {
return buscalls.Fill{}, err
}
if name, err = js.StreamNameBySubject(ctx, h.prefix+".>"); err != nil {
return buscalls.Fill{}, fmt.Errorf("no stream on the bus keeps %s: %w", h.prefix, err)
}
}
f, err := h.read.streamFill(name)
return f.Fill, err
}
// purgeSubject purges one subject's messages from the stream that keeps it: the stream named, or found on the runner's
// own connection by its subject, purged on the ledger connection.
func purgeSubject(read, write *Conn, stream, subject string) error {
if write == nil {
return errors.New("a ledger's streams are purged only on its module's ledger connection, and none is held")
}
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
defer cancel()
if stream == "" {
js, err := read.JetStream()
if err != nil {
return err
}
if stream, err = js.StreamNameBySubject(ctx, subject); err != nil {
return fmt.Errorf("no stream on the bus keeps %s: %w", subject, err)
}
}
// The purge is asked of the stream by its name, with no read of its info first: the ledger account is granted the
// purge of its own ledger's two streams and nothing else (PurgeAPI).
body, err := json.Marshal(map[string]string{"filter": subject})
if err != nil {
return err
}
reply, err := write.nc.RequestWithContext(ctx, PurgeAPI(stream), body)
if err != nil {
return fmt.Errorf("purging %s from %s: %w", subject, stream, err)
}
var ack struct {
Success bool `json:"success"`
Error *struct {
Description string `json:"description"`
} `json:"error"`
}
if err := json.Unmarshal(reply.Data, &ack); err != nil || !ack.Success {
why := string(reply.Data)
if ack.Error != nil {
why = ack.Error.Description
}
return fmt.Errorf("the bus refused the purge of %s from %s: %s", subject, stream, why)
}
return nil
}
// PurgeAPI is the bus's subject that purges a stream, the one a ledger account is granted for its own ledger's streams.
func PurgeAPI(stream string) string { return "$JS.API.STREAM.PURGE." + stream }
type busState struct {
c *Conn
bucket string
@@ -231,6 +302,11 @@ func (s *busSnapshots) Get(key string) ([]byte, uint64, error) {
return lastOn(s.read, "", s.prefix+"."+key)
}
// Purge removes one instance's snapshot, on the module's ledger connection.
func (s *busSnapshots) Purge(key string) error {
return purgeSubject(s.read, s.write, "", s.prefix+"."+key)
}
func (s *busSnapshots) Put(key string, value []byte, expect uint64) error {
if s.write == nil {
return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held")
+79
View File
@@ -58,6 +58,10 @@ type History interface {
// Read is the operations of the instances keys names (a key, with `*` for one name and `>` for the rest) from a
// sequence on, oldest first, at most limit of them (every one when limit is 0).
Read(keys string, from uint64, limit int) ([]Kept, error)
// Purge removes every operation of the instance key names, and nothing of another (an instance's removal).
Purge(key string) error
// Fill is the history's fill against its cap, the ledger's cap (buscalls.LedgerFill; design 47 G §587).
Fill() (buscalls.Fill, error)
}
// State is a ledger's state bucket: each instance's record under its key.
@@ -69,6 +73,8 @@ type State interface {
Write(key string, value []byte, revision uint64) (uint64, error)
// Keys is every key the bucket holds.
Keys() ([]string, error)
// Delete removes the key; a key it does not hold is no fault.
Delete(key string) error
}
// Snapshots are a ledger's snapshots: the latest record of each instance, on a subject of its own, never expiring.
@@ -77,6 +83,8 @@ type Snapshots interface {
Get(key string) ([]byte, uint64, error)
// Put writes the key's snapshot, expecting its last to be at expect (0: none); ErrRaced when it is not.
Put(key string, value []byte, expect uint64) error
// Purge removes the key's snapshot; one it does not hold is no fault.
Purge(key string) error
}
// Ledger is one ledger of one module: its types and where it is kept.
@@ -495,3 +503,74 @@ func (l *Ledger) Apply(r buscalls.ApplyRequest, author buscalls.Author) (buscall
}
return buscalls.Applied{Taken: true, Sequence: seq, StateBehind: behind}, nil
}
// Fill is the fill of the ledger's history against the ledger's cap (buscalls.LedgerFill; design 47 G §587): its messages,
// its bytes and the cap in bytes. The history holds every operation, and the cap is its alone: neither the fullest of the
// stores nor their sum is answered. A history that cannot say its fill is an error naming the ledger, never a fill of
// nothing.
func (l *Ledger) Fill(buscalls.LedgerRequest) (buscalls.Fill, error) {
f, err := l.History.Fill()
if err != nil {
return buscalls.Fill{}, fmt.Errorf("%s's ledger %s cannot say the fill of its history: %w", l.Module, l.Name, err)
}
return f, nil
}
// Remove removes one instance with everything it kept (buscalls.LedgerRemove; design 47 G §587, §155): its operations, its
// snapshot, its key in the state bucket and its keys in every declared index, and answers how many operations and how many
// bytes of the history that was. An instance the ledger does not hold answers 0 and 0, and a removal that left something
// (said, and answered as an error) is made again by calling it again, which removes the rest.
func (l *Ledger) Remove(r buscalls.LedgerRemoveRequest) (buscalls.LedgerRemoved, error) {
key, err := KeyOf(r.Instance)
if err != nil {
return buscalls.LedgerRemoved{}, err
}
out := buscalls.LedgerRemoved{Instance: r.Instance}
before, err := l.History.Fill()
if err != nil {
return out, fmt.Errorf("%s's ledger %s cannot say the fill of its history, so a removal is not made: %w", l.Module, l.Name, err)
}
ops, err := l.History.Read(key, 1, 0)
if err != nil {
return out, err
}
for _, step := range []struct {
what string
remove func(string) error
}{{"operations", l.History.Purge}, {"snapshot", l.Snapshots.Purge}, {"state", l.State.Delete}, {"index entries", l.dropFromIndexes}} {
if err := step.remove(key); err != nil {
l.say("[mesh-tools] %s's ledger %s removes %s and its %s could not be removed: %v; removing it again removes what "+
"is left", l.Module, l.Name, r.Instance, step.what, err)
return out, fmt.Errorf("%s's ledger %s removed part of %s and its %s are left: %w", l.Module, l.Name, r.Instance, step.what, err)
}
}
after, err := l.History.Fill()
if err != nil {
return out, fmt.Errorf("%s's ledger %s removed %s and cannot say the fill of its history after: %w", l.Module, l.Name, r.Instance, err)
}
out.Operations = uint64(len(ops))
if before.Bytes > after.Bytes {
out.Bytes = before.Bytes - after.Bytes
}
return out, nil
}
// dropFromIndexes removes the entries of the instance key names from every index bucket the ledger is bound to.
func (l *Ledger) dropFromIndexes(key string) error {
instance := InstanceOf(key)
for name, ix := range l.Indexes {
held, err := ix.Entries("", 0)
if err != nil {
return fmt.Errorf("the index %s: %w", name, err)
}
for _, e := range held {
var v buscalls.IndexEntry
if json.Unmarshal(e.Value, &v) == nil && v.Instance == instance {
if err := ix.Delete(e.Key); err != nil {
return fmt.Errorf("the index %s: %w", name, err)
}
}
}
}
return nil
}
+47
View File
@@ -100,3 +100,50 @@ func TestApplyAppendsDerivesAndSnapshots(t *testing.T) {
})
}
}
// design 47 G §587: a ledger's fill is its history's against the ledger's cap, a store that cannot say it is an error naming
// the ledger; a removal takes an instance's operations, snapshot, state and index entries and answers what that was, an
// unknown instance answers none, and nothing of another instance goes.
func TestTheLedgerSaysItsFillAndRemovesAnInstanceWhole(t *testing.T) {
l, f := newLedger()
f.history.MaxBytes = 1 << 20
for _, n := range []string{"a", "b"} {
applied(t, l, "decision/0290", "note", `{"says":"`+n+`"}`, owner)
}
applied(t, l, "decision/0291", "note", `{"says":"other"}`, owner)
entry, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0290", Sequence: 2})
other, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0291", Sequence: 1})
_ = f.index.Put("t.decision.0290", entry)
_ = f.index.Put("t.decision.0291", other)
fill, err := l.Fill(buscalls.LedgerRequest{Ledger: "records"})
if err != nil || fill.Count != 3 || fill.Bytes == 0 || fill.MaxBytes != 1<<20 {
t.Fatalf("the fill is %+v (%v), want 3 messages against 1 MiB", fill, err)
}
got, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"})
if err != nil || got.Instance != "decision/0290" || got.Operations != 2 || got.Bytes == 0 {
t.Fatalf("the removal answered %+v (%v), want 2 operations and their bytes", got, err)
}
if kept, _ := f.history.Read("decision.0290", 1, 0); len(kept) != 0 {
t.Errorf("operations are left: %d", len(kept))
}
if raw, _, _ := f.state.Get("decision.0290"); raw != nil {
t.Errorf("the state is left: %s", raw)
}
if raw, _, _ := f.snapshots.Get("decision.0290"); raw != nil {
t.Errorf("the snapshot is left: %s", raw)
}
if held, _ := f.index.Entries("", 0); len(held) != 1 || held[0].Key != "t.decision.0291" {
t.Errorf("the index holds %v, want the other instance's entry alone", held)
}
if kept, _ := f.history.Read("decision.0291", 1, 0); len(kept) != 1 {
t.Errorf("another instance lost operations: %d left", len(kept))
}
if again, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}); err != nil || again.Operations != 0 || again.Bytes != 0 {
t.Errorf("a removal said twice answered %+v (%v), want none", again, err)
}
f.history.FillFails = true
if _, err := l.Fill(buscalls.LedgerRequest{}); err == nil || !strings.Contains(err.Error(), "ledger records cannot say the fill") {
t.Errorf("a history that cannot say its fill answered %v", err)
}
}
+50 -3
View File
@@ -1,9 +1,12 @@
package ledger
import (
"errors"
"sort"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/buscalls"
)
// The memory stores keep what the bus keeps, as the bus answers it, for a test of a ledger without a bus: a stream whose
@@ -12,9 +15,13 @@ import (
// MemoryHistory is a history in memory.
type MemoryHistory struct {
mu sync.Mutex
seq uint64
kept []struct {
// MaxBytes is the cap its fill is said against.
MaxBytes uint64
// FillFails makes Fill an error, as a store that cannot say its fill.
FillFails bool
mu sync.Mutex
seq uint64
kept []struct {
key string
Kept
}
@@ -73,6 +80,32 @@ func (h *MemoryHistory) Read(keys string, from uint64, limit int) ([]Kept, error
return out, nil
}
func (h *MemoryHistory) Purge(key string) error {
h.mu.Lock()
defer h.mu.Unlock()
kept := h.kept[:0]
for _, k := range h.kept {
if k.key != key {
kept = append(kept, k)
}
}
h.kept = kept
return nil
}
func (h *MemoryHistory) Fill() (buscalls.Fill, error) {
h.mu.Lock()
defer h.mu.Unlock()
if h.FillFails {
return buscalls.Fill{}, errors.New("the history does not say its fill")
}
f := buscalls.Fill{Count: uint64(len(h.kept)), MaxBytes: h.MaxBytes}
for _, k := range h.kept {
f.Bytes += uint64(len(k.Data))
}
return f, nil
}
// Expire drops every operation of the history, as a ledger's keeping days end them.
func (h *MemoryHistory) Expire() {
h.mu.Lock()
@@ -129,6 +162,13 @@ func (s *MemoryState) Keys() ([]string, error) {
return out, nil
}
func (s *MemoryState) Delete(key string) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.kv, key)
return nil
}
// MemorySnapshots are snapshots in memory.
type MemorySnapshots struct {
mu sync.Mutex
@@ -157,6 +197,13 @@ func (s *MemorySnapshots) Put(key string, value []byte, expect uint64) error {
return nil
}
func (s *MemorySnapshots) Purge(key string) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.kv, key)
return nil
}
// MemoryIndex is an index's bucket in memory.
type MemoryIndex struct {
mu sync.Mutex
+51 -9
View File
@@ -3,6 +3,8 @@ package runtime
import (
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"os"
"slices"
"sort"
"strings"
"testing"
@@ -140,20 +142,60 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) {
t.Fatalf("ping of the instance: %v", err)
}
var ping micro.Ping
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" {
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err)
}
}
// Design 47 G §462: the call that reads a sent object is listed under the bus state capability, and the ledger calls keep theirs.
func TestTheSentObjectCallBelongsToTheBusStateCapability(t *testing.T) {
for c, method := range map[string]string{CapBusStateCalls: buscalls.ObjectsSentGet} {
found := false
for _, m := range callFamilies[c] {
found = found || m == method
// Design 47 G §462, §488, §622: each call is stated under the capability word of its own family, the words mesh-sdk's buscalls
// names: the read of a sent object under sent-objects-calls (no longer under bus-state-calls), the ledger's fill and removal
// under ledger-upkeep-calls, the others where they were.
func TestEachCallIsStatedUnderTheWordOfItsFamily(t *testing.T) {
for method, word := range map[string]string{
buscalls.ObjectsSentGet: buscalls.SentObjectsCalls,
buscalls.LedgerFill: buscalls.LedgerUpkeepCalls,
buscalls.LedgerRemove: buscalls.LedgerUpkeepCalls,
buscalls.LedgerApply: buscalls.LedgerCalls,
buscalls.ObjectsPut: buscalls.BusStateCalls,
} {
var in []string
for c, methods := range callFamilies {
if slices.Contains(methods, method) {
in = append(in, c)
}
}
if !found {
t.Errorf("%s does not list %s", c, method)
if len(in) != 1 || in[0] != word {
t.Errorf("%s is stated under %v, want %s alone", method, in, word)
}
}
}
// Design 47 G §545: the words the tool runner states are exactly its manifest's artifact `states`, so the delivery plan's edge
// of §605 runs to it for each word a module's artifact `calls`.
func TestTheWordsTheRunnerStatesAreItsManifestsStates(t *testing.T) {
raw, err := os.ReadFile("../../module.json")
if err != nil {
t.Fatal(err)
}
var m struct {
Build struct {
Artifacts []struct {
Name string `json:"name"`
States []string `json:"states"`
} `json:"artifacts"`
} `json:"build"`
}
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
var declared []string
for _, a := range m.Build.Artifacts {
if a.Name == "runtime" {
declared = a.States
}
}
slices.Sort(declared)
if got := CallCapabilities(); !slices.Equal(got, declared) {
t.Errorf("the runner states %v, its manifest's states are %v", got, declared)
}
}
+11 -5
View File
@@ -18,8 +18,12 @@ import (
// (announce.Capabilities) for its machine's node-engine to take into the node's record. A module whose manifest asks for one is
// held off a machine whose runner does not state it.
const (
CapBusStateCalls = "bus-state-calls"
CapLedgerCalls = "ledger-calls"
CapBusStateCalls = buscalls.BusStateCalls
CapLedgerCalls = buscalls.LedgerCalls
// CapLedgerUpkeepCalls is the ledger's fill and removal (§462, §488) and CapSentObjectsCalls the read of a sent object (§488):
// each is stated once the runner answers its family, the words mesh-sdk's buscalls names (§622).
CapLedgerUpkeepCalls = buscalls.LedgerUpkeepCalls
CapSentObjectsCalls = buscalls.SentObjectsCalls
// CapPutSessions is the put and get sessions' word (design 47 G §518, §622), mesh-sdk's.
CapPutSessions = stdio.CapPutSessions
)
@@ -27,10 +31,12 @@ const (
// callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer.
var callFamilies = map[string][]string{
CapBusStateCalls: {buscalls.ObjectsPut, buscalls.ObjectsGet, buscalls.ObjectsList, buscalls.ObjectsDelete, buscalls.ObjectsInfo,
buscalls.ObjectsWatch, buscalls.ObjectsSentGet, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead},
buscalls.ObjectsWatch, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead},
CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery,
buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag},
CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk},
CapLedgerUpkeepCalls: {buscalls.LedgerFill, buscalls.LedgerRemove},
CapSentObjectsCalls: {buscalls.ObjectsSentGet},
CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk},
}
// CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat
@@ -111,7 +117,7 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes
answer, err = conn.WorkQueueDead(b.module, r)
}
case buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerQuery, buscalls.LedgerRebuild,
buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag:
buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag, buscalls.LedgerFill, buscalls.LedgerRemove:
answer, err = b.ledgerCall(method, params)
default:
return nil, fmt.Errorf("the runtime answers no %s", method)
@@ -240,6 +240,35 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) {
t.Fatalf("queried %+v, %v", q, err)
}
}},
{"fill and removal are answered through the calls and remove an instance from the bus whole", "decision.gone", func(t *testing.T) {
must(t, open(t), "decision/gone", "title", `{"title":"Gone"}`)
must(t, open(t), "decision/gone", "note", `{"says":"two"}`)
must(t, open(t), "decision/stays", "note", `{"says":"kept"}`)
raw, err := m.BusState(buscalls.LedgerFill, json.RawMessage(`{"ledger":"records"}`))
var fill buscalls.Fill
if _ = json.Unmarshal(raw, &fill); err != nil || fill.Count < 3 || fill.Bytes == 0 {
t.Fatalf("the fill answered %s (%v)", raw, err)
}
raw, err = m.BusState(buscalls.LedgerRemove, json.RawMessage(`{"ledger":"records","instance":"decision/gone"}`))
var removed buscalls.LedgerRemoved
if _ = json.Unmarshal(raw, &removed); err != nil || removed.Operations != 2 || removed.Bytes == 0 {
t.Fatalf("the removal answered %s (%v)", raw, err)
}
l := open(t)
if last, _ := l.History.Last("decision.gone"); last != 0 {
t.Fatalf("operations are left, at %d", last)
}
if st, _, _ := l.State.Get("decision.gone"); st != nil {
t.Fatalf("the state is left: %s", st)
}
if last, _ := l.History.Last("decision.stays"); last == 0 {
t.Fatal("another instance lost its operations")
}
if raw, err = m.BusState(buscalls.LedgerRemove, json.RawMessage(`{"ledger":"records","instance":"decision/gone"}`)); err != nil ||
!strings.Contains(string(raw), `"operations":0`) {
t.Fatalf("a removal said twice answered %s (%v)", raw, err)
}
}},
{"a forged author is refused", "decision.forged", func(t *testing.T) {
_, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"decision/forged","operation":"title","body":{"title":"T","author":"op@shanks"},"call":"forged"}`))
if err == nil || !strings.Contains(err.Error(), "the call forged, which is not a tool call in flight") {
+11
View File
@@ -171,6 +171,17 @@ func (b *moduleBus) ledgerCall(method string, params json.RawMessage) (any, erro
var r buscalls.LedgerRequest
_ = json.Unmarshal(params, &r)
return l.Lag(r)
case buscalls.LedgerFill:
var r buscalls.LedgerRequest
_ = json.Unmarshal(params, &r)
return l.Fill(r)
case buscalls.LedgerRemove:
// Never on an apply: a removal is no operation of the history and has no author (design 47 G §587).
var r buscalls.LedgerRemoveRequest
if err := decode(method, params, &r); err != nil {
return nil, err
}
return l.Remove(r)
}
return nil, fmt.Errorf("the runtime answers no %s", method)
}
+1 -1
View File
@@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) {
// no other name; none answers "the runtime answers no ...".
func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) {
m, _, _ := aLedger(t)
if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" {
if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
t.Fatalf("the runner states %s", got)
}
for capability, methods := range callFamilies {
+9 -1
View File
@@ -62,7 +62,15 @@
"language": "go",
"system": "arch",
"from": "cmd/node-tools",
"binary": "node-tools"
"binary": "node-tools",
"states": [
"bus-state-calls",
"ledger-calls",
"ledger-upkeep-calls",
"put-sessions",
"seat-settle-delay",
"sent-objects-calls"
]
}
]
},