novox/mesh-sdk
9
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Revert G's mesh-sdk merge (#189): G merged half, and today's validator refuses G's delivery parts
mesh/delivery merged: source.main moved to novox/mesh-sdk@27da949140c6
Details
mesh/check being carried to 8afc3ebdf371 on 93e0a7b50d90 | from 27da949140c6

Browse Source 
The group feat/47-G-join11 merged 3 of 17 members on 2026-11-13; G's mesh-sdk alone on main breaks 25 trunk
modules. The operator chose to put the order right: a reader release first, then G whole. This restores main
to 8fbba810 exactly.

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-13 11:12:48 +01:00
parent b9915abf51
commit 8afc3ebdf3
212 changed files with 2436 additions and 32088 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/stateobjects/bundle.go
-78
View File
@@ -1,78 +0,0 @@
package stateobjects
import (
"encoding/json"
"errors"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/busstate"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// ForState is a state of this module, kept through its runtime: the state by its local name, and its object store
// `<state>-objects` (catalogue.StateObjectsName).
func ForState(state string) Values {
store := catalogue.StateObjectsName(state)
return Values{KV: BundleKV{State: stdio.State(state)}, Objects: BundleObjects{Store: busstate.ObjectsOf(store)}, Store: store}
}
// BundleKV is a state as a bundle reaches it.
type BundleKV struct{ State stdio.KeptState }
func (b BundleKV) Get(key string) (*Entry, error) {
e, err := b.State.Get(key)
if err != nil || e == nil {
return nil, err
}
return &Entry{Value: e.Value, Revision: e.Revision}, nil
}
func changed(err error) error {
if errors.Is(err, stdio.ErrChanged) {
return ErrChanged
}
return err
}
func (b BundleKV) Create(key string, value []byte) error {
_, err := b.State.Create(key, json.RawMessage(value))
return changed(err)
}
func (b BundleKV) Update(key string, value []byte, revision uint64) error {
_, err := b.State.Update(key, json.RawMessage(value), revision)
return changed(err)
}
func (b BundleKV) Delete(key string) error { return b.State.Delete(key) }
func (b BundleKV) Keys() ([]string, error) { return b.State.Keys() }
// BundleObjects is an object store as a bundle reaches it.
type BundleObjects struct{ Store busstate.Objects }
// MostBytes is zero: a bundle's value of any size is put by a session of chunks, bounded by its store's cap, which the tool
// runner refuses at the chunk that passes it (design 47 G §518).
func (BundleObjects) MostBytes() int { return 0 }
func (b BundleObjects) Put(name string, data []byte) error {
_, err := b.Store.Put(name, data, "a value kept beside its key")
return err
}
func (b BundleObjects) Get(name string) ([]byte, error) {
d, err := b.Store.Get(name)
return d.Data, err
}
func (b BundleObjects) Delete(name string) error { return b.Store.Delete(name) }
func (b BundleObjects) List() ([]Object, error) {
l, err := b.Store.List()
out := make([]Object, 0, len(l))
for _, o := range l {
if !o.Deleted {
out = append(out, Object{Name: o.Name, Size: int64(o.Size), Modified: o.Modified})
}
}
return out, err
}
go/stateobjects/fakes_test.go
-274
View File
@@ -1,274 +0,0 @@
package stateobjects
import (
"bufio"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"sort"
"strings"
"sync"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/buscalls"
)
// memory is a state and its object store in memory, behind the two interfaces.
type memory struct {
mu sync.Mutex
rev uint64
keys map[string]Entry
objects map[string][]byte
failPuts bool
modified map[string]time.Time
most int
puts map[string][]byte // a put session's chunks so far, by call
}
func newMemory() *memory { return &memory{keys: map[string]Entry{}, objects: map[string][]byte{}} }
func (m *memory) Get(key string) (*Entry, error) {
m.mu.Lock()
defer m.mu.Unlock()
if e, ok := m.keys[key]; ok {
return &e, nil
}
return nil, nil
}
func (m *memory) write(key string, value []byte, revision *uint64) error {
m.mu.Lock()
defer m.mu.Unlock()
e, has := m.keys[key]
if (revision == nil && has) || (revision != nil && (!has || e.Revision != *revision)) {
return ErrChanged
}
m.rev++
m.keys[key] = Entry{Value: value, Revision: m.rev}
return nil
}
func (m *memory) Create(key string, value []byte) error { return m.write(key, value, nil) }
func (m *memory) Update(key string, value []byte, revision uint64) error {
return m.write(key, value, &revision)
}
func (m *memory) Delete(key string) error {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.keys, key)
return nil
}
func (m *memory) Keys() ([]string, error) {
m.mu.Lock()
defer m.mu.Unlock()
var out []string
for k := range m.keys {
out = append(out, k)
}
sort.Strings(out)
return out, nil
}
func (m *memory) Put(name string, data []byte) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.failPuts {
return errors.New("the store is full")
}
m.objects[name] = data
if m.modified != nil {
if _, ok := m.modified[name]; !ok {
m.modified[name] = time.Time{}
}
}
return nil
}
func (m *memory) GetObject(name string) ([]byte, error) { return m.Object(name) }
func (m *memory) values() Values { return Values{KV: m, Objects: memObjects{m}, Store: "s-objects"} }
// Object is an object's bytes.
func (m *memory) Object(name string) ([]byte, error) {
m.mu.Lock()
defer m.mu.Unlock()
d, ok := m.objects[name]
if !ok {
return nil, fmt.Errorf("no object %s", name)
}
return d, nil
}
// memObjects is the memory's object store side (Put is the memory's own).
type memObjects struct{ m *memory }
func (o memObjects) MostBytes() int { return o.m.most }
func (o memObjects) Put(name string, data []byte) error { return o.m.Put(name, data) }
func (o memObjects) Get(name string) ([]byte, error) { return o.m.Object(name) }
func (o memObjects) Delete(name string) error {
o.m.mu.Lock()
defer o.m.mu.Unlock()
delete(o.m.objects, name)
return nil
}
func (o memObjects) List() ([]Object, error) {
o.m.mu.Lock()
defer o.m.mu.Unlock()
var out []Object
for n, d := range o.m.objects {
out = append(out, Object{Name: n, Size: int64(len(d)), Modified: o.m.modified[n]})
}
return out, nil
}
// runtimeOf serves one bundle tool, "run", against the memory as the tool runner would answer the state's and the object
// store's calls, and returns what the tool answered. The tool runs run in the bundle, which reaches the memory through
// stdio.State and busstate.Objects.
func runtimeOf(t *testing.T, m *memory, run func() (any, error)) any {
t.Helper()
t.Setenv(stdio.ServedAs, "owner")
inR, inW := io.Pipe()
outR, outW := io.Pipe()
go func() {
_ = stdio.ServeOn("", []stdio.Tool{{Name: "run", Run: func(map[string]any) (any, error) { return run() }}}, inR, outW)
}()
var wmu sync.Mutex
send := func(v map[string]any) {
v["jsonrpc"] = "2.0"
b, _ := json.Marshal(v)
wmu.Lock()
defer wmu.Unlock()
_, _ = inW.Write(append(b, '\n'))
}
answered := make(chan any, 1)
go func() {
lines := bufio.NewScanner(outR)
lines.Buffer(make([]byte, 1<<20), 1<<28)
for lines.Scan() {
var in struct {
ID json.RawMessage `json:"id"`
Method string `json:"method"`
Params json.RawMessage `json:"params"`
Result json.RawMessage `json:"result"`
}
if json.Unmarshal(lines.Bytes(), &in) != nil {
continue
}
if !strings.HasPrefix(in.Method, "mesh/") {
if string(in.ID) == "1" {
answered <- string(in.Result)
return
}
continue
}
go func() {
res, err := m.serve(in.Method, in.Params)
a := map[string]any{"id": json.RawMessage(in.ID)}
if err != nil {
code := -32000
if errors.Is(err, ErrChanged) {
code = stdio.CodeChanged
}
a["error"] = map[string]any{"code": code, "message": err.Error()}
} else {
a["result"] = res
}
send(a)
}()
}
}()
send(map[string]any{"id": 1, "method": "tools/call", "params": map[string]any{"name": "run"}})
select {
case a := <-answered:
return a
case <-time.After(20 * time.Second):
t.Fatal("the bundle said nothing")
return nil
}
}
func (m *memory) serve(method string, raw json.RawMessage) (any, error) {
var p struct {
Key string `json:"key"`
Value json.RawMessage `json:"value"`
Revision uint64 `json:"revision"`
Name string `json:"name"`
Data []byte `json:"data"`
Object string `json:"object"`
}
_ = json.Unmarshal(raw, &p)
switch method {
case "mesh/state.get":
e, _ := m.Get(p.Key)
if e == nil {
return nil, nil
}
return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value), "revision": e.Revision}, nil
case "mesh/state.create":
return map[string]any{"revision": 1}, m.Create(p.Key, p.Value)
case "mesh/state.update":
return map[string]any{"revision": 1}, m.Update(p.Key, p.Value, p.Revision)
case "mesh/state.delete":
return buscalls.Done{}, m.Delete(p.Key)
case "mesh/state.keys":
return m.Keys()
case stdio.CapabilitiesCall:
return map[string]any{"words": []string{stdio.CapPutSessions}}, nil
case buscalls.ObjectsPut:
return buscalls.ObjectInfo{Name: p.Name, Size: uint64(len(p.Data))}, m.Put(p.Name, p.Data)
case buscalls.ObjectsPutChunk:
// a put session (design 47 G §518): the chunks are kept by call and the object is put on the last
var c buscalls.ObjectsPutChunkRequest
_ = json.Unmarshal(raw, &c)
m.mu.Lock()
if m.puts == nil {
m.puts = map[string][]byte{}
}
m.puts[c.Call] = append(m.puts[c.Call], c.Data...)
whole := m.puts[c.Call]
if c.Last {
delete(m.puts, c.Call)
}
m.mu.Unlock()
if !c.Last {
return buscalls.ObjectsChunkTaken{Seq: c.Seq}, nil
}
return buscalls.ObjectInfo{Name: c.Name, Size: uint64(len(whole))}, m.Put(c.Name, whole)
case buscalls.ObjectsGet, buscalls.ObjectsSentGet:
name := p.Name
if method == buscalls.ObjectsSentGet {
name = p.Object
}
d, err := m.Object(name)
if err == nil && len(d) > buscalls.ChunkBytes {
sum := sha256.Sum256(d)
return map[string]any{"chunked": (len(d) + buscalls.ChunkBytes - 1) / buscalls.ChunkBytes, "size": len(d),
"digest": "sha256:" + hex.EncodeToString(sum[:]), "call": name}, nil
}
return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: name}, Data: d}, err
case buscalls.ObjectsGetChunk:
var c buscalls.ObjectsGetChunkRequest
_ = json.Unmarshal(raw, &c)
d, err := m.Object(c.Call)
at := int(c.Seq-1) * buscalls.ChunkBytes
end := min(at+buscalls.ChunkBytes, len(d))
return buscalls.ObjectsChunk{Seq: c.Seq, Last: end == len(d), Data: d[at:end]}, err
case buscalls.ObjectsDelete:
return buscalls.Done{}, memObjects{m}.Delete(p.Name)
case buscalls.ObjectsList:
l, _ := memObjects{m}.List()
var out []buscalls.ObjectInfo
for _, o := range l {
out = append(out, buscalls.ObjectInfo{Name: o.Name, Size: uint64(o.Size)})
}
return buscalls.ObjectsListed{Objects: out}, nil
}
return nil, fmt.Errorf("no %s", method)
}
go/stateobjects/process.go
-126
View File
@@ -1,126 +0,0 @@
package stateobjects
import (
"context"
"errors"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
)
// callTimeout bounds each call a process makes of its bus.
const callTimeout = 30 * time.Second
// OverBus is a state a process reaches through its own JetStream, with the object store beside it: a foundation part's
// bucket and its old `objects` section, or a bundle-less part's `<state>-objects`. store names the object store as a
// reader asks for it, written into every reference. The bucket's own limit for one value is the limit of the values.
func OverBus(kv jetstream.KeyValue, objects jetstream.ObjectStore, store string) Values {
limit, cancel := context.WithTimeout(context.Background(), callTimeout)
defer cancel()
return Values{KV: BusKV{KV: kv}, Objects: BusObjects{Store: objects}, Store: store, Limit: int(kvlimit.LimitOf(limit, kv))}
}
// BusKV is a bucket as a process reaches it.
type BusKV struct{ KV jetstream.KeyValue }
func ctx() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), callTimeout)
}
func (b BusKV) Get(key string) (*Entry, error) {
c, cancel := ctx()
defer cancel()
e, err := b.KV.Get(c, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
if e.Operation() != jetstream.KeyValuePut {
return nil, nil
}
return &Entry{Value: e.Value(), Revision: e.Revision()}, nil
}
func (b BusKV) Create(key string, value []byte) error {
c, cancel := ctx()
defer cancel()
_, err := kvlimit.Create(c, b.KV, key, value)
if errors.Is(err, jetstream.ErrKeyExists) {
return ErrChanged
}
return err
}
func (b BusKV) Update(key string, value []byte, revision uint64) error {
c, cancel := ctx()
defer cancel()
_, err := kvlimit.Update(c, b.KV, key, value, revision)
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
return ErrChanged
}
return err
}
func (b BusKV) Delete(key string) error {
c, cancel := ctx()
defer cancel()
return b.KV.Delete(c, key)
}
func (b BusKV) Keys() ([]string, error) {
c, cancel := ctx()
defer cancel()
keys, err := b.KV.Keys(c)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
return keys, err
}
// BusObjects is an object store as a process reaches it; it takes any size in one put.
type BusObjects struct{ Store jetstream.ObjectStore }
func (BusObjects) MostBytes() int { return 0 }
func (b BusObjects) Put(name string, data []byte) error {
c, cancel := ctx()
defer cancel()
_, err := b.Store.PutBytes(c, name, data)
return err
}
func (b BusObjects) Get(name string) ([]byte, error) {
c, cancel := ctx()
defer cancel()
return b.Store.GetBytes(c, name)
}
func (b BusObjects) Delete(name string) error {
c, cancel := ctx()
defer cancel()
err := b.Store.Delete(c, name)
if errors.Is(err, jetstream.ErrObjectNotFound) {
return nil
}
return err
}
func (b BusObjects) List() ([]Object, error) {
c, cancel := ctx()
defer cancel()
l, err := b.Store.List(c)
if errors.Is(err, jetstream.ErrNoObjectsFound) {
return nil, nil
}
out := make([]Object, 0, len(l))
for _, o := range l {
if !o.Deleted {
out = append(out, Object{Name: o.Name, Size: int64(o.Size), Modified: o.ModTime})
}
}
return out, err
}
go/stateobjects/ref.go
-71
View File
@@ -1,71 +0,0 @@
// Package stateobjects keeps values of any size on the bus (novox/hq ADR 0666 §16, ADR 0595 §9): a value over the bus's
// limit for one value, or a verb reply or an event over the limit for one message, is put in an object store and only its
// reference travels. It is a small package of its own, imported only by the modules that use it: the module library (the
// root package) and the process package do not import it, so their importers are not rebuilt when it changes. It names no
// seat and no module, and imports no contracts module.
package stateobjects
import (
"bytes"
"encoding/hex"
"encoding/json"
"fmt"
)
// FormVersion is the version of the reference form; a reader refuses a reference of another.
const FormVersion = 1
// refField is the one field of a body that is a reference.
const refField = "mesh-object"
// Ref names an object in an object store: a body that is exactly `{"mesh-object": Ref}` is a reference, and anything
// else is a value.
type Ref struct {
// Store is the object store's name as its reader asks for it: a bundle's local name, or the name a process opens.
Store string `json:"store"`
// Object is the object's name in that store.
Object string `json:"object"`
// Size is the object's size in bytes.
Size int64 `json:"size"`
// Digest is the hex SHA-256 of the object's bytes; a reader checks it after the read.
Digest string `json:"sha256"`
// Form is FormVersion.
Form int `json:"form"`
}
func (r Ref) valid() bool {
if r.Store == "" || r.Object == "" || r.Size < 0 || r.Form != FormVersion || len(r.Digest) != 64 {
return false
}
_, err := hex.DecodeString(r.Digest)
return err == nil
}
// Wrap is the body that holds r.
func Wrap(r Ref) []byte {
raw, _ := json.Marshal(map[string]Ref{refField: r})
return raw
}
// ReadRef reads a body as a reference: ok is false for a value, including one that only looks like a reference (another
// field beside it, a field it does not know, a digest that is no digest, another form).
func ReadRef(body []byte) (Ref, bool) {
var outer map[string]json.RawMessage
if err := json.Unmarshal(body, &outer); err != nil || len(outer) != 1 {
return Ref{}, false
}
raw, ok := outer[refField]
if !ok {
return Ref{}, false
}
dec := json.NewDecoder(bytes.NewReader(raw))
dec.DisallowUnknownFields()
var r Ref
if err := dec.Decode(&r); err != nil || !r.valid() {
return Ref{}, false
}
return r, true
}
// errLooksLikeRef refuses a value that would be read back as a reference.
var errLooksLikeRef = fmt.Errorf("the value is exactly a reference to an object (a body of the one field %q): it would be read as one", refField)
go/stateobjects/ref_test.go
-43
View File
@@ -1,43 +0,0 @@
package stateobjects
import (
"strings"
"testing"
)
// ADR 0666 §16: a body is a reference when it is exactly `{"mesh-object": Ref}`; anything else is a value.
func subABodyIsAReferenceOnlyWhenItIsExactlyOne(t *testing.T) {
good := Ref{Store: "views-objects", Object: "k." + strings.Repeat("a", 64), Size: 3 << 20, Digest: strings.Repeat("a", 64), Form: FormVersion}
got, ok := ReadRef(Wrap(good))
if !ok || got != good {
t.Fatalf("round trip: %+v, %v", got, ok)
}
d := strings.Repeat("a", 64)
rows := []struct{ name, body string }{
{"a value", `{"a":1}`},
{"a list", `[1]`},
{"not JSON", `nope`},
{"another field beside it", `{"mesh-object":{"store":"s","object":"o","size":1,"sha256":"` + d + `","form":1},"x":1}`},
{"a field the reference does not know", `{"mesh-object":{"store":"s","object":"o","size":1,"sha256":"` + d + `","form":1,"x":1}}`},
{"a digest that is no digest", `{"mesh-object":{"store":"s","object":"o","size":1,"sha256":"zz","form":1}}`},
{"another form", `{"mesh-object":{"store":"s","object":"o","size":1,"sha256":"` + d + `","form":2}}`},
{"no store", `{"mesh-object":{"object":"o","size":1,"sha256":"` + d + `","form":1}}`},
}
for _, r := range rows {
if _, ok := ReadRef([]byte(r.body)); ok {
t.Errorf("%s: read as a reference", r.name)
}
}
}
// TestABodyIsAReferenceOnlyWhenItIsExactlyOne runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
func TestABodyIsAReferenceOnlyWhenItIsExactlyOne(t *testing.T) {
t.Run("ABodyIsAReferenceOnlyWhenItIsExactlyOne", subABodyIsAReferenceOnlyWhenItIsExactlyOne)
t.Run("ABodyOverTheMessageLimitIsSentAsAReference", subABodyOverTheMessageLimitIsSentAsAReference)
t.Run("ABundleReadsASentObjectThroughTheRuntime", subABundleReadsASentObjectThroughTheRuntime)
t.Run("AFoundationPartsSentStoreAgesFromItsSendLog", subAFoundationPartsSentStoreAgesFromItsSendLog)
t.Run("TheObjectToolsReadEverythingAndDropOnlyWhatIsNotLive", subTheObjectToolsReadEverythingAndDropOnlyWhatIsNotLive)
t.Run("ValuesOfAnySizeThroughTheRuntime", subValuesOfAnySizeThroughTheRuntime)
t.Run("AWriteLostLeavesNoObjectBehind", subAWriteLostLeavesNoObjectBehind)
t.Run("AValueLargerThanTheStoreTakesIsRefused", subAValueLargerThanTheStoreTakesIsRefused)
}
go/stateobjects/send.go
-159
View File
@@ -1,159 +0,0 @@
package stateobjects
import (
"encoding/json"
"errors"
"fmt"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/busstate"
)
// MessageLimit is the most a message of the bus carries when the sender does not know its connection's own
// (`nc.MaxPayload()`), and MessageHeaderRoom what the headers and the subject take of it.
const (
MessageLimit = 1 << 20
MessageHeaderRoom = 8 << 10
)
// Sender sends the body of a verb's reply or an event: a body over the per-message limit is put in the sender's store of
// sent objects and its reference is sent instead (ADR 0666 §16). The store is the bus module's `<module>_sent-objects`,
// granted to the sender alone and read by the callers its seat definition names and the consumers that declare the event;
// a sent object lives as long as that store's age, which is transport and not kept data.
type Sender struct {
// Objects is the sender's store of sent objects, and Store its name as a reader asks for it.
Objects Objects
Store string
// Limit is the connection's largest message; zero is MessageLimit.
Limit int
// Log is a foundation part's old `state` section `sent-log`, with Node the key of this node's sends in it and Now
// the clock (zero is time.Now): the sender keeps the sequence of its own sends there and ages the store from it at
// each send (design 47 G §410, §330). A sender with no Log ages nothing: the bus expires a template store's objects.
Log KV
Node string
Now func() time.Time
}
// The sent store of a foundation part keeps an object SentMaxAge, and a send deletes at most SentAgeAtMost of the
// objects past it (design 47 G §330, §59).
const (
SentLogState = "sent-log"
SentAgeAtMost = 20
)
type sentEntry struct {
Object string `json:"object"`
At time.Time `json:"at"`
}
// record writes this send into the sequence of the sender's sends by compare-and-set, and deletes the objects of the
// oldest sends that are past SentMaxAge, at most SentAgeAtMost; two sends at once lose none, the one that lost the
// compare reading the sequence again.
func (s Sender) record(object string) error {
now := time.Now
if s.Now != nil {
now = s.Now
}
for {
e, err := s.Log.Get(s.Node)
if err != nil {
return err
}
var log []sentEntry
if e != nil {
if err := json.Unmarshal(e.Value, &log); err != nil {
return fmt.Errorf("the %s of %s: %w", SentLogState, s.Node, err)
}
}
var aged []sentEntry
for len(aged) < SentAgeAtMost && len(log) > 0 && now().Sub(log[0].At) > SentMaxAge {
aged, log = append(aged, log[0]), log[1:]
}
log = append(log, sentEntry{object, now()})
raw, _ := json.Marshal(log)
if e == nil {
err = s.Log.Create(s.Node, raw)
} else {
err = s.Log.Update(s.Node, raw, e.Revision)
}
if errors.Is(err, ErrChanged) {
continue
}
if err != nil {
return err
}
for _, a := range aged {
if err := s.Objects.Delete(a.Object); err != nil {
return fmt.Errorf("aging the sent object %s: %w", a.Object, err)
}
}
return nil
}
}
// Send is what to send for body: the body itself when it fits, else the reference of the object it was put in as
// `<label>.<sha256>`. label tells the object's use (the verb or event, the caller or the id).
func (s Sender) Send(label string, body []byte) ([]byte, error) {
limit := s.Limit
if limit <= 0 {
limit = MessageLimit
}
if len(body) <= limit-MessageHeaderRoom {
if _, isRef := ReadRef(body); isRef {
return nil, fmt.Errorf("%s: %w", label, errLooksLikeRef)
}
return body, nil
}
if most := s.Objects.MostBytes(); most > 0 && len(body) > most {
return nil, fmt.Errorf("%s: a bundle sends at most %d bytes of body, the runtime's channel carrying a message of 16 MiB; this one is %d", label, most, len(body))
}
ref := Ref{Store: s.Store, Object: objectName(label, digestOf(body)), Size: int64(len(body)), Digest: digestOf(body), Form: FormVersion}
if err := s.Objects.Put(ref.Object, body); err != nil {
// A send refused at its store's cap fails loudly, naming the store (design 47 G §477, §59).
return nil, fmt.Errorf("the object for %s in the sent store %s: %w", label, s.Store, err)
}
if s.Log != nil {
if err := s.record(ref.Object); err != nil {
return nil, fmt.Errorf("the %s for %s: %w", SentLogState, label, err)
}
}
return Wrap(ref), nil
}
// Fetch reads the object a reference names.
type Fetch func(Ref) ([]byte, error)
// Receive is the body that was sent: the body itself, or the object its reference names, fetched and checked against the
// reference's size and digest.
func Receive(body []byte, fetch Fetch) ([]byte, error) {
ref, ok := ReadRef(body)
if !ok {
return body, nil
}
data, err := fetch(ref)
if err != nil {
return nil, fmt.Errorf("the object %s of %s: %w", ref.Object, ref.Store, err)
}
if int64(len(data)) != ref.Size || digestOf(data) != ref.Digest {
return nil, fmt.Errorf("the object %s of %s is not the one its reference names (size or digest differ)", ref.Object, ref.Store)
}
return data, nil
}
// BundleFetch reads a sent object through the runtime, which serves it under the standing grant the grants part derives
// from the manifests (buscalls.ObjectsSentGet).
func BundleFetch(ref Ref) ([]byte, error) {
raw, err := stdio.BusState(buscalls.ObjectsSentGet, buscalls.ObjectsSentGetRequest{Store: ref.Store, Object: ref.Object, Size: ref.Size, Digest: ref.Digest})
if err != nil {
return nil, err
}
d, err := busstate.ReadObject(ref.Object, raw)
return d.Data, err
}
// ProcessFetch reads a sent object through the reader's own object store.
func ProcessFetch(o Objects) Fetch {
return func(ref Ref) ([]byte, error) { return o.Get(ref.Object) }
}
go/stateobjects/send_test.go
-96
View File
@@ -1,96 +0,0 @@
package stateobjects
import (
"bytes"
"strings"
"testing"
"time"
)
// ADR 0666 §16: a reply or an event over the per-message limit carries the name of an object, and the receiver reads it
// back whole; one that fits is sent as it is.
func subABodyOverTheMessageLimitIsSentAsAReference(t *testing.T) {
m := newMemory()
s := Sender{Objects: memObjects{m}, Store: "m_sent-objects"}
fetch := ProcessFetch(memObjects{m})
for _, label := range []string{"reply.list", "event.done"} {
body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
sent, err := s.Send(label, body)
if err != nil || len(sent) > MessageLimit {
t.Fatalf("%s: sent %d bytes, %v", label, len(sent), err)
}
if _, isRef := ReadRef(sent); !isRef {
t.Fatalf("%s: a 3 MiB body was sent whole", label)
}
got, err := Receive(sent, fetch)
if err != nil || !bytes.Equal(got, body) {
t.Fatalf("%s: received %d bytes, %v", label, len(got), err)
}
}
small := []byte(`{"a":1}`)
if sent, err := s.Send("x", small); err != nil || !bytes.Equal(sent, small) {
t.Fatalf("a small body: %s, %v", sent, err)
}
if got, err := Receive(small, fetch); err != nil || !bytes.Equal(got, small) {
t.Fatalf("receive a small body: %s, %v", got, err)
}
// an object that is not the one the reference names is refused
sent, _ := s.Send("reply.again", bytes.Repeat([]byte("z"), 2<<20))
ref, _ := ReadRef(sent)
m.objects[ref.Object] = []byte("tampered")
if _, err := Receive(sent, fetch); err == nil {
t.Fatal("a tampered object was accepted")
}
// a bundle's store is bounded by its cap alone, and the tool runner refuses at the chunk that passes it (design 47 G §518)
if (BundleObjects{}).MostBytes() != 0 {
t.Fatal("a bundle's value is refused by a number of its own")
}
}
// A bundle reads a sent object through the runtime's sent-get.
func subABundleReadsASentObjectThroughTheRuntime(t *testing.T) {
m := newMemory()
body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
sent, err := Sender{Objects: memObjects{m}, Store: "m_sent-objects"}.Send("reply.list", body)
if err != nil {
t.Fatal(err)
}
out := runtimeOf(t, m, func() (any, error) {
got, err := Receive(sent, BundleFetch)
if err != nil || !bytes.Equal(got, body) {
t.Errorf("received %d bytes, %v", len(got), err)
}
return "done", nil
})
if !strings.Contains(out.(string), "done") {
t.Fatalf("the bundle answered %v", out)
}
}
// Design 47 G §410, §330: a foundation part's sender keeps the sequence of its sends in its old state section
// `sent-log` and deletes the sent objects past 7 days at a later send, at most 20 a send.
func subAFoundationPartsSentStoreAgesFromItsSendLog(t *testing.T) {
m := newMemory()
clock := time.Date(2026, 11, 1, 0, 0, 0, 0, time.UTC)
s := Sender{Objects: memObjects{m}, Store: "m_sent-objects", Log: m, Node: "n1", Now: func() time.Time { return clock }}
var first string
for i := 0; i < 25; i++ {
sent, err := s.Send("reply."+string(rune('a'+i)), bytes.Repeat([]byte{byte('a' + i)}, 2<<20))
if err != nil {
t.Fatal(err)
}
if ref, _ := ReadRef(sent); i == 0 {
first = ref.Object
}
}
if len(m.objects) != 25 {
t.Fatalf("%d objects within 7 days, want 25", len(m.objects))
}
clock = clock.Add(SentMaxAge + time.Hour)
if _, err := s.Send("reply.late", bytes.Repeat([]byte("z"), 2<<20)); err != nil {
t.Fatal(err)
}
if _, kept := m.objects[first]; kept || len(m.objects) != 6 {
t.Fatalf("after a late send %d objects remain (the first kept: %v), want 6", len(m.objects), kept)
}
}
go/stateobjects/tools.go
-258
View File
@@ -1,258 +0,0 @@
package stateobjects
import (
"errors"
"fmt"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/views"
)
// SentMaxAge is how long a store of sent objects keeps one: transport, not kept data (ADR 0595 §4), so the store ages
// its objects out by itself and the drop tool is for the case where it does not.
const SentMaxAge = 7 * 24 * time.Hour
// State is one state of a module with its objects: the state's name (its local name in the manifest), its values and the
// fill of its object store against its cap.
type State struct {
Name string
Values Values
Fill func() (buscalls.Fill, error)
}
// Sent is a module's store of sent objects, read and dropped by the module's own tools.
type Sent struct {
Objects Objects
Fill func() (buscalls.Fill, error)
// MaxAge is the store's age (SentMaxAge when zero), which the listing compares each object's age with.
MaxAge time.Duration
}
// Tools are the four tools of a module that keeps large values or sends large bodies (design 25 § 12.2, ADR 0666 §25),
// named for the module the caller gives: `<module>_state_objects` and `<module>_state_objects_drop` over the states, and
// `<module>_sent_objects` and `<module>_sent_objects_drop` over the store of sent objects. The two reading tools read
// everything the store holds; the two dropping tools remove only what is no longer live, each only with the person who
// asks (`by`) and why, recorded as a hand act before anything is removed, and refuse one that cannot be recorded.
func Tools(module string, states []State, sent *Sent, acts views.HandActs, now func() time.Time) []stdio.Tool {
var tools []stdio.Tool
if len(states) > 0 {
tools = append(tools, stateObjectsTool(module, states), stateDropTool(module, states, acts))
}
if sent != nil {
tools = append(tools, sentObjectsTool(module, *sent, now), sentDropTool(module, *sent, acts, now))
}
return tools
}
func pick(states []State, args map[string]any) (State, error) {
name, _ := args["state"].(string)
name = strings.TrimSpace(name)
if name == "" && len(states) == 1 {
return states[0], nil
}
for _, s := range states {
if s.Name == name {
return s, nil
}
}
var names []string
for _, s := range states {
names = append(names, s.Name)
}
return State{}, fmt.Errorf("state names one of %s, not %q", strings.Join(names, ", "), name)
}
var stateArg = map[string]any{"type": "string", "description": "the state whose object store is read: its local name in the manifest"}
func stateObjectsTool(module string, states []State) stdio.Tool {
return stdio.Tool{
Name: module + "_state_objects",
Description: "The objects that hold the large values of " + module + "'s states: each object, its size, the key whose value names it " +
"or key-gone when the key was deleted (such an object stays until " + module + "_state_objects_drop removes it on a person's word), " +
"and how full the object store is against its cap.",
Input: map[string]any{"type": "object", "properties": map[string]any{"state": stateArg}},
Run: func(args map[string]any) (any, error) {
s, err := pick(states, args)
if err != nil {
return nil, err
}
kept, err := s.Values.Kept()
if err != nil {
return nil, err
}
out := map[string]any{"state": s.Name, "objects": kept}
gone := 0
for _, k := range kept {
if k.KeyGone {
gone++
}
}
out["key-gone"] = gone
if s.Fill != nil {
if f, err := s.Fill(); err == nil {
out["fill"] = f
} else {
out["fill-error"] = err.Error()
}
}
return out, nil
},
}
}
var byWhy = map[string]any{
"by": map[string]any{"type": "string", "description": "who asks, as who@node"},
"why": map[string]any{"type": "string", "description": "why, in plain words"},
}
// asked is by and why of a dropping call: both are needed, or nothing is done.
func asked(args map[string]any) (by, why string, err error) {
by, _ = args["by"].(string)
why, _ = args["why"].(string)
by, why = strings.TrimSpace(by), strings.TrimSpace(why)
if by == "" || why == "" {
return "", "", errors.New("a drop names who asks (by) and why: nothing was dropped")
}
return by, why, nil
}
func record(acts views.HandActs, tool, what, by, why string) error {
if acts == nil {
return errors.New("no way to record a hand act is given, so nothing was dropped")
}
if err := acts.RecordHandAct(views.HandAct{What: what + " (asked by " + by + ")", Why: why, Cause: "objects-dropped", Verb: tool}); err != nil {
return fmt.Errorf("the hand act was not recorded, so nothing was dropped: %w", err)
}
return nil
}
func stateDropTool(module string, states []State, acts views.HandActs) stdio.Tool {
name := module + "_state_objects_drop"
props := map[string]any{"state": stateArg}
for k, v := range byWhy {
props[k] = v
}
return stdio.Tool{
Name: name,
Description: "Drop the objects of one of " + module + "'s states whose key is gone, on a person's word (ADR 0595 §4, ADR 0666 §16). " +
"An object a key names is never dropped. Recorded as a hand act first; needs by and why.",
Input: map[string]any{"type": "object", "required": []string{"by", "why"}, "properties": props},
Run: func(args map[string]any) (any, error) {
by, why, err := asked(args)
if err != nil {
return nil, err
}
s, err := pick(states, args)
if err != nil {
return nil, err
}
if err := record(acts, name, "dropped the objects of "+module+"'s state "+s.Name+" whose key is gone", by, why); err != nil {
return nil, err
}
dropped, err := s.Values.DropKeyGone()
return map[string]any{"state": s.Name, "dropped": dropped}, err
},
}
}
// SentObject is one object of a store of sent objects as the listing shows it.
type SentObject struct {
Name string `json:"name"`
// Label is the verb or event the body was sent for.
Label string `json:"label"`
Size int64 `json:"size"`
Sent time.Time `json:"sent,omitzero"`
Age string `json:"age"`
PastAge bool `json:"past-age,omitempty"`
}
func (s Sent) maxAge() time.Duration {
if s.MaxAge > 0 {
return s.MaxAge
}
return SentMaxAge
}
func (s Sent) list(now time.Time) ([]SentObject, error) {
objects, err := s.Objects.List()
if err != nil {
return nil, err
}
out := make([]SentObject, 0, len(objects))
for _, o := range objects {
age := now.Sub(o.Modified)
out = append(out, SentObject{Name: o.Name, Label: keyOf(o.Name), Size: o.Size, Sent: o.Modified,
Age: age.Round(time.Second).String(), PastAge: age > s.maxAge()})
}
return out, nil
}
func sentObjectsTool(module string, s Sent, now func() time.Time) stdio.Tool {
return stdio.Tool{
Name: module + "_sent_objects",
Description: "The objects " + module + " sent in place of a reply or an event over the message limit: each object, its size, when it " +
"was sent, the verb or event it was sent for, its age against the store's age (" + s.maxAge().String() + ", transport and not kept data), " +
"and how full the store is against its cap.",
Run: func(map[string]any) (any, error) {
objects, err := s.list(now())
if err != nil {
return nil, err
}
out := map[string]any{"objects": objects, "max-age": s.maxAge().String()}
if s.Fill != nil {
if f, err := s.Fill(); err == nil {
out["fill"] = f
} else {
out["fill-error"] = err.Error()
}
}
return out, nil
},
}
}
func sentDropTool(module string, s Sent, acts views.HandActs, now func() time.Time) stdio.Tool {
name := module + "_sent_objects_drop"
props := map[string]any{"older-than": map[string]any{"type": "string", "description": "a duration such as 24h: only objects sent longer ago are dropped"}}
for k, v := range byWhy {
props[k] = v
}
return stdio.Tool{
Name: name,
Description: "Drop the sent objects of " + module + " older than a given age, on a person's word. Younger objects are live and stay. " +
"Recorded as a hand act first; needs by, why and older-than.",
Input: map[string]any{"type": "object", "required": []string{"by", "why", "older-than"}, "properties": props},
Run: func(args map[string]any) (any, error) {
by, why, err := asked(args)
if err != nil {
return nil, err
}
text, _ := args["older-than"].(string)
age, err := time.ParseDuration(strings.TrimSpace(text))
if err != nil || age <= 0 {
return nil, fmt.Errorf("older-than %q is not a duration above zero such as 24h: nothing was dropped", text)
}
if err := record(acts, name, "dropped the sent objects of "+module+" older than "+age.String(), by, why); err != nil {
return nil, err
}
objects, err := s.Objects.List()
if err != nil {
return nil, err
}
var dropped []string
for _, o := range objects {
if now().Sub(o.Modified) <= age {
continue
}
if err := s.Objects.Delete(o.Name); err != nil {
return map[string]any{"dropped": dropped}, fmt.Errorf("the object %s: %w", o.Name, err)
}
dropped = append(dropped, o.Name)
}
return map[string]any{"dropped": dropped}, nil
},
}
}
go/stateobjects/tools_test.go
-110
View File
@@ -1,110 +0,0 @@
package stateobjects
import (
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
"git.novox.be/novox/mesh-sdk/go/views"
)
type acts struct {
done []views.HandAct
err error
}
func (a *acts) RecordHandAct(h views.HandAct) error {
if a.err != nil {
return a.err
}
a.done = append(a.done, h)
return nil
}
// ADR 0666 §25: the four tools read what each store holds and drop only what is no longer live, each drop only with who
// asks and why, recorded as a hand act first.
func subTheObjectToolsReadEverythingAndDropOnlyWhatIsNotLive(t *testing.T) {
m := newMemory()
m.modified = map[string]time.Time{}
v := m.values()
big := big(2 * kvlimit.ModuleStateValueBytes)
for _, k := range []string{"live", "gone"} {
if err := v.Put(k, big); err != nil {
t.Fatal(err)
}
}
// the two keys hold one object each (different keys): delete one key
if err := v.Delete("gone"); err != nil {
t.Fatal(err)
}
now := time.Date(2026, 11, 10, 12, 0, 0, 0, time.UTC)
rec := &acts{}
sentStore := newMemory()
sentStore.modified = map[string]time.Time{"reply.list.aa": now.Add(-48 * time.Hour), "event.done.bb": now.Add(-time.Hour)}
sentStore.objects = map[string][]byte{"reply.list.aa": []byte("1"), "event.done.bb": []byte("22")}
tools := Tools("m", []State{{Name: "views", Values: v}}, &Sent{Objects: memObjects{sentStore}}, rec, func() time.Time { return now })
by := map[string]func(map[string]any) (any, error){}
for _, tl := range tools {
by[tl.Name] = tl.Run
}
for _, name := range []string{"m_state_objects", "m_state_objects_drop", "m_sent_objects", "m_sent_objects_drop"} {
if by[name] == nil {
t.Fatalf("no tool %s", name)
}
}
read, err := by["m_state_objects"](map[string]any{})
if err != nil || read.(map[string]any)["key-gone"] != 1 {
t.Fatalf("the state's objects: %v, %v", read, err)
}
listed, err := by["m_sent_objects"](map[string]any{})
if objs := listed.(map[string]any)["objects"].([]SentObject); err != nil || len(objs) != 2 {
t.Fatalf("the sent objects: %v, %v", listed, err)
}
// each drop refuses without by and why, and records nothing
for _, tc := range []struct {
tool string
args map[string]any
}{
{"m_state_objects_drop", map[string]any{"why": "x"}},
{"m_state_objects_drop", map[string]any{"by": "a@b"}},
{"m_sent_objects_drop", map[string]any{"why": "x", "older-than": "24h"}},
{"m_sent_objects_drop", map[string]any{"by": "a@b", "why": "x", "older-than": "soon"}},
} {
if _, err := by[tc.tool](tc.args); err == nil {
t.Errorf("%s %v was accepted", tc.tool, tc.args)
}
}
if len(rec.done) != 0 {
t.Fatalf("a refused drop recorded %v", rec.done)
}
// the hand act is recorded first: one that cannot be recorded drops nothing
rec.err = errors.New("hand acts are down")
if _, err := by["m_state_objects_drop"](map[string]any{"by": "a@b", "why": "tidy"}); err == nil || !strings.Contains(err.Error(), "nothing was dropped") {
t.Fatalf("an unrecorded drop: %v", err)
}
if kept, _ := v.Kept(); len(kept) != 2 {
t.Fatalf("an unrecorded drop removed objects: %d left", len(kept))
}
rec.err = nil
// the state drop takes only the key-gone object
out, err := by["m_state_objects_drop"](map[string]any{"by": "a@b", "why": "tidy"})
if err != nil || len(out.(map[string]any)["dropped"].([]string)) != 1 {
t.Fatalf("the state drop: %v, %v", out, err)
}
if kept, _ := v.Kept(); len(kept) != 1 || kept[0].KeyGone {
t.Fatalf("the live object did not survive: %+v", kept)
}
// the sent drop takes only objects past the age
out, err = by["m_sent_objects_drop"](map[string]any{"by": "a@b", "why": "tidy", "older-than": "24h"})
if err != nil || len(out.(map[string]any)["dropped"].([]string)) != 1 {
t.Fatalf("the sent drop: %v, %v", out, err)
}
if _, ok := sentStore.objects["event.done.bb"]; !ok || len(sentStore.objects) != 1 {
t.Fatalf("the young sent object did not survive: %v", sentStore.objects)
}
if len(rec.done) != 2 {
t.Fatalf("%d hand acts, want 2", len(rec.done))
}
}
go/stateobjects/values.go
-240
View File
@@ -1,240 +0,0 @@
package stateobjects
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
)
// ErrChanged is a write refused because the key was written by another writer first; nothing was written.
var ErrChanged = errors.New("the key was written by another writer first")
// Entry is a key's current value and its revision.
type Entry struct {
Value []byte
Revision uint64
}
// KV is the state's key-value side, as a bundle reaches it through its runtime or a process through its own JetStream.
type KV interface {
// Get is the key's current entry, or nil when it has none.
Get(key string) (*Entry, error)
// Create writes the key only when it has none, Update only while it is at revision; either refuses as ErrChanged.
Create(key string, value []byte) error
Update(key string, value []byte, revision uint64) error
Delete(key string) error
// Keys is every key with a value.
Keys() ([]string, error)
}
// Object is one object an object store holds.
type Object struct {
Name string
Size int64
Modified time.Time
}
// Objects is the object store beside the state.
type Objects interface {
// MostBytes is the largest object it takes in one put, or zero for no limit: a value over it is refused, loudly.
MostBytes() int
Put(name string, data []byte) error
Get(name string) ([]byte, error)
Delete(name string) error
List() ([]Object, error)
}
// Values is a state whose values may be of any size: a value that fits the bucket is kept in its key, a larger one in
// the object store beside it with its reference in the key (ADR 0666 §16). Store is the object store's name as a reader
// asks for it, written into every reference.
type Values struct {
KV KV
Objects Objects
Store string
// Limit is the most a key's value may hold, headers included; zero is kvlimit.ModuleStateValueBytes.
Limit int
}
func (v Values) inlineMost() int {
limit := v.Limit
if limit <= 0 {
limit = kvlimit.ModuleStateValueBytes
}
return limit - kvlimit.HeaderRoom
}
func digestOf(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
// objectName is a large value's object, `<key>.<digest>`: the same value is the same object, a changed one another.
func objectName(key, digest string) string { return key + "." + digest }
// Put writes the value as JSON under key: inline up to the limit, else as an object `<key>.<sha256>` with its reference
// in the key. A replace leaves the object it replaced, as a past value may still name it (ADR 0595 §4): it is listed
// key-gone until the drop tool removes it on a person's word. A write another writer won first is ErrChanged and leaves
// what it found. A value that is itself exactly a reference is refused.
func (v Values) Put(key string, value any) error {
data, err := json.Marshal(value)
if err != nil {
return fmt.Errorf("the value for %s: %w", key, err)
}
if _, isRef := ReadRef(data); isRef {
return fmt.Errorf("%s: %w", key, errLooksLikeRef)
}
old, err := v.KV.Get(key)
if err != nil {
return err
}
var oldRef Ref
var hadRef bool
if old != nil {
oldRef, hadRef = ReadRef(old.Value)
}
body, object := data, ""
if len(data) > v.inlineMost() {
ref := Ref{Store: v.Store, Object: objectName(key, digestOf(data)), Size: int64(len(data)), Digest: digestOf(data), Form: FormVersion}
object = ref.Object
if most := v.Objects.MostBytes(); most > 0 && len(data) > most {
return fmt.Errorf("the value for %s is %d bytes, over the %d an object may hold through this runtime: it is refused", key, len(data), most)
}
if !(hadRef && oldRef.Object == object) {
if err := v.Objects.Put(object, data); err != nil {
return fmt.Errorf("the object for %s: %w", key, err)
}
}
body = Wrap(ref)
}
if old == nil {
err = v.KV.Create(key, body)
} else {
err = v.KV.Update(key, body, old.Revision)
}
if err != nil {
if object != "" && !(hadRef && oldRef.Object == object) {
// an object no key and no past value names: made for this write alone
_ = v.Objects.Delete(object)
}
return err
}
return nil
}
// Get reads the key's value into into and says whether the key has one. A value kept as an object is fetched and
// checked against its reference's size and digest.
func (v Values) Get(key string, into any) (bool, error) {
e, err := v.KV.Get(key)
if err != nil || e == nil {
return false, err
}
data := e.Value
if ref, ok := ReadRef(e.Value); ok {
if data, err = v.Objects.Get(ref.Object); err != nil {
return false, fmt.Errorf("the object %s of %s: %w", ref.Object, key, err)
}
if int64(len(data)) != ref.Size || digestOf(data) != ref.Digest {
return false, fmt.Errorf("the object %s of %s is not the one its reference names (size or digest differ)", ref.Object, key)
}
}
return true, json.Unmarshal(data, into)
}
// Delete removes the key. Its object stays, listed as key-gone, until the drop tool removes it on a person's word
// (ADR 0666 §16, ADR 0595 §4).
func (v Values) Delete(key string) error { return v.KV.Delete(key) }
// Kept is one object of a state's store as the reading tool shows it.
type Kept struct {
Name string `json:"name"`
Size int64 `json:"size"`
Modified time.Time `json:"modified,omitzero"`
// Key is the key whose value names it; for an object whose key is gone, the key it was written for.
Key string `json:"key"`
// KeyGone says no key names the object any longer.
KeyGone bool `json:"key-gone,omitempty"`
}
// Kept lists every object of the store with the key naming it, or key-gone, by name.
func (v Values) Kept() ([]Kept, error) {
live, err := v.live()
if err != nil {
return nil, err
}
objects, err := v.Objects.List()
if err != nil {
return nil, err
}
out := make([]Kept, 0, len(objects))
for _, o := range objects {
k := Kept{Name: o.Name, Size: o.Size, Modified: o.Modified}
if key, ok := live[o.Name]; ok {
k.Key = key
} else {
k.KeyGone = true
k.Key = keyOf(o.Name)
}
out = append(out, k)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out, nil
}
// live is the objects some key names now, by object name to key.
func (v Values) live() (map[string]string, error) {
keys, err := v.KV.Keys()
if err != nil {
return nil, err
}
live := map[string]string{}
for _, key := range keys {
e, err := v.KV.Get(key)
if err != nil {
return nil, err
}
if e == nil {
continue
}
if ref, ok := ReadRef(e.Value); ok {
live[ref.Object] = key
}
}
return live, nil
}
// DropKeyGone removes the objects whose key is gone, and only those, answering their names. It is the act of the drop
// tool; a live object, one a key names, is never removed.
func (v Values) DropKeyGone() ([]string, error) {
kept, err := v.Kept()
if err != nil {
return nil, err
}
var dropped []string
for _, k := range kept {
if !k.KeyGone {
continue
}
if err := v.Objects.Delete(k.Name); err != nil {
return dropped, fmt.Errorf("the object %s: %w", k.Name, err)
}
dropped = append(dropped, k.Name)
}
return dropped, nil
}
// keyOf is the key an object `<key>.<digest>` was written for.
func keyOf(object string) string {
if i := strings.LastIndex(object, "."); i > 0 {
if _, err := hex.DecodeString(object[i+1:]); err == nil && len(object)-i-1 == 64 {
return object[:i]
}
}
return object
}
go/stateobjects/values_integration_test.go
-98
View File
@@ -1,98 +0,0 @@
//go:build integration
package stateobjects
import (
"bytes"
"context"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/testbus"
"git.novox.be/novox/mesh-sdk/go/views"
)
// ADR 0666 §16, §25 against the real bus: a process's values of 1x, 10x and 100x the limit, replaced and deleted, and a
// body over the per-message limit sent and received, each read back whole.
func TestStateObjectsWholeAgainstTheBus(t *testing.T) {
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "m_views", MaxValueSize: 256 * 1024, Storage: jetstream.MemoryStorage})
if err != nil {
t.Fatal(err)
}
store, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_views-objects", Storage: jetstream.MemoryStorage})
if err != nil {
t.Fatal(err)
}
sent, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_sent-objects", Storage: jetstream.MemoryStorage, TTL: 7 * 24 * time.Hour})
if err != nil {
t.Fatal(err)
}
t.Run("a reply and an event over the message limit", func(t *testing.T) {
s := Sender{Objects: BusObjects{Store: sent}, Store: "m_sent-objects", Limit: int(nc.MaxPayload())}
for _, label := range []string{"reply.list", "event.done"} {
body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
wire, err := s.Send(label, body)
if err != nil || len(wire) > int(nc.MaxPayload()) {
t.Fatalf("%s: sent %d bytes, %v", label, len(wire), err)
}
got, err := Receive(wire, ProcessFetch(BusObjects{Store: sent}))
if err != nil || !bytes.Equal(got, body) {
t.Fatalf("%s: received %d bytes, %v", label, len(got), err)
}
}
})
// G §477: a send past a fixture's cap fails its reply or event, naming the store.
t.Run("a send past the sent store's cap names the store", func(t *testing.T) {
small, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "n_sent-objects", Storage: jetstream.MemoryStorage, MaxBytes: 1 << 20})
if err != nil {
t.Fatal(err)
}
s := Sender{Objects: BusObjects{Store: small}, Store: "n_sent-objects", Limit: int(nc.MaxPayload())}
if _, err := s.Send("reply.list", []byte(`{"text":"`+strings.Repeat("y", 2<<20)+`"}`)); err == nil || !strings.Contains(err.Error(), "n_sent-objects") {
t.Fatalf("a send past the cap: %v, want it refused naming n_sent-objects", err)
}
})
t.Run("values of 1x, 10x and 100x the limit", func(t *testing.T) { valueRows(t, OverBus(kv, store, "m_views-objects"), 1, 10, 100) })
t.Run("the tools read and drop what a deleted key left", func(t *testing.T) {
v := OverBus(kv, store, "m_views-objects")
if err := v.Put("tool-key", big(2*256*1024)); err != nil {
t.Fatal(err)
}
if err := v.Delete("tool-key"); err != nil {
t.Fatal(err)
}
tools := Tools("m", []State{{Name: "views", Values: v}}, nil, handActs{}, time.Now)
read, err := tools[0].Run(map[string]any{})
if err != nil || read.(map[string]any)["key-gone"].(int) < 1 {
t.Fatalf("the reading tool: %v, %v", read, err)
}
if _, err := tools[1].Run(map[string]any{"by": "a@b", "why": "test"}); err != nil {
t.Fatal(err)
}
kept, _ := v.Kept()
for _, k := range kept {
if k.KeyGone {
t.Fatalf("%s is still key-gone after the drop", k.Name)
}
}
})
}
type handActs struct{}
func (handActs) RecordHandAct(views.HandAct) error { return nil }
go/stateobjects/values_test.go
-172
View File
@@ -1,172 +0,0 @@
package stateobjects
import (
"bytes"
"encoding/json"
"slices"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/kvlimit"
)
// big is a JSON value of about n bytes.
func big(n int) map[string]string { return map[string]string{"text": strings.Repeat("x", n)} }
// valueRows is what every backend of Values holds: a value of 1x, 10x and 100x the limit is read back whole, a replace
// leaves the old object and a delete leaves its object, both listed key-gone until the drop (ADR 0666 §16, ADR 0595 §4).
func valueRows(t *testing.T, v Values, multiples ...int) {
t.Helper()
limit := kvlimit.ModuleStateValueBytes
for _, times := range multiples {
key := map[int]string{1: "one", 10: "ten", 100: "hundred"}[times]
want := big(times * limit)
if err := v.Put(key, want); err != nil {
t.Errorf("%dx: put: %v", times, err)
return
}
var got map[string]string
if ok, err := v.Get(key, &got); err != nil || !ok || got["text"] != want["text"] {
t.Errorf("%dx: read back %v, found %v, %v", times, len(got["text"]), ok, err)
return
}
e, _ := v.KV.Get(key)
if _, isRef := ReadRef(e.Value); !isRef {
t.Errorf("%dx: the key holds the value, not a reference", times)
return
}
}
if err := v.Put("small", map[string]int{"a": 1}); err != nil {
t.Error(err)
return
}
if e, _ := v.KV.Get("small"); !bytes.Equal(e.Value, []byte(`{"a":1}`)) {
t.Errorf("a small value is kept in its key, got %s", e.Value)
return
}
// a replace leaves the object it replaced, listed key-gone (ADR 0595 §4); the same value again adds none
before, _ := v.Objects.List()
if err := v.Put("one", big(2*limit)); err != nil {
t.Error(err)
return
}
if err := v.Put("one", big(2*limit)); err != nil {
t.Error(err)
return
}
after, _ := v.Objects.List()
if len(after) != len(before)+1 {
t.Errorf("a replace left %d objects, %d before it: the old one must stay", len(after), len(before))
return
}
if err := v.Put("one", "small now"); err != nil {
t.Error(err)
return
}
if now, _ := v.Objects.List(); len(now) != len(after) {
t.Errorf("replacing by a small value changed the objects: %d, want %d", len(now), len(after))
return
}
// a delete leaves the object, listed key-gone, and the drop removes it alone
if err := v.Delete("ten"); err != nil {
t.Error(err)
return
}
kept, err := v.Kept()
if err != nil {
t.Error(err)
return
}
var gone []string
for _, k := range kept {
if k.KeyGone {
gone = append(gone, k.Key)
}
}
slices.Sort(gone)
if !slices.Equal(gone, []string{"one", "one", "ten"}) {
t.Errorf("the objects whose key is gone: %v, want the two replaced of one and ten's", gone)
return
}
dropped, err := v.DropKeyGone()
if err != nil || len(dropped) != 3 {
t.Errorf("dropped %v, %v", dropped, err)
return
}
if left, _ := v.Objects.List(); len(left) != len(kept)-3 {
t.Errorf("the drop left %d objects of %d", len(left), len(kept))
return
}
if err := v.Put("ref", json.RawMessage(Wrap(Ref{Store: "s", Object: "o", Digest: strings.Repeat("a", 64), Form: FormVersion}))); err == nil {
t.Error("a value that is exactly a reference was accepted")
return
}
}
// ADR 0666 §16, C-S3, C-S5: values of 1x, 10x and 100x the limit through a bundle's runtime, whole.
func subValuesOfAnySizeThroughTheRuntime(t *testing.T) {
m := newMemory()
out := runtimeOf(t, m, func() (any, error) {
v := ForState("views")
valueRows(t, v, 1, 10, 100) // a bundle puts a value of any size by session (design 47 G §518)
return "done", nil
})
if !strings.Contains(out.(string), "done") {
t.Fatalf("the bundle answered %v", out)
}
}
// A write another writer won first leaves the object it put removed and the key as it found it.
func subAWriteLostLeavesNoObjectBehind(t *testing.T) {
m := newMemory()
v := m.values()
if err := v.Put("k", big(3*kvlimit.ModuleStateValueBytes)); err != nil {
t.Fatal(err)
}
// another writer changes the key between this writer's read and write
racing := &raceKV{KV: m, then: func() { _ = m.Update("k", []byte(`"other"`), 1) }}
lost := Values{KV: racing, Objects: memObjects{m}, Store: "s"}
if err := lost.Put("k", big(4*kvlimit.ModuleStateValueBytes)); err == nil || err != ErrChanged {
t.Fatalf("the lost write: %v", err)
}
if l, _ := (memObjects{m}).List(); len(l) != 1 {
t.Fatalf("%d objects, want the first value's only", len(l))
}
m.failPuts = true
if err := v.Put("k2", big(3*kvlimit.ModuleStateValueBytes)); err == nil {
t.Fatal("a put into a full store was accepted")
}
if e, _ := m.Get("k2"); e != nil {
t.Fatal("the key was written though its object was not")
}
}
// raceKV runs then once, after the first Get, as another writer would.
type raceKV struct {
KV
then func()
}
func (r *raceKV) Get(key string) (*Entry, error) {
e, err := r.KV.Get(key)
if r.then != nil {
then := r.then
r.then = nil
then()
}
return e, err
}
// A store that takes no object as large as the value refuses it, loudly, and writes nothing.
func subAValueLargerThanTheStoreTakesIsRefused(t *testing.T) {
m := newMemory()
m.most = 100_000
v := m.values()
err := v.Put("k", big(5*kvlimit.ModuleStateValueBytes))
if err == nil || !strings.Contains(err.Error(), "refused") {
t.Fatalf("the put: %v", err)
}
if e, _ := m.Get("k"); e != nil || len(m.objects) != 0 {
t.Fatal("a refused value left a key or an object")
}
}