From 2dbcc89ed244d2a79ddd742b86d5acc23d8079ca Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 10:36:59 +0100 Subject: [PATCH 001/134] D-1: the forms of a caller entry and the default callers of a verb's class MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Text callers are still read here: G's reader refuses them only once the bootstrapped copies are lists (D-4 to D-6). Builds: design 47 G §336, §392, §391, §68, §338 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/seatdef/callers.go | 22 ++++++++++++++++++++++ go/seatdef/seatdef_test.go | 19 +++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/go/seatdef/callers.go b/go/seatdef/callers.go index 177e4c16..ba60222d 100644 --- a/go/seatdef/callers.go +++ b/go/seatdef/callers.go @@ -101,6 +101,28 @@ func CallerEntryProblem(e string) string { "holder:@same-node", e) } +// DefaultCallers are the callers a verb that names none takes by its class (design 47 G §391): a read takes invokers, +// persons and sessions; an act the operator and sessions; a warrant the operator. A verb that names no class is a +// read. +func DefaultCallers(class string) []string { + switch class { + case ClassAct: + return []string{CallerOperator, CallerSessions} + case ClassWarrant: + return []string{CallerOperator} + } + return []string{CallerInvokers, CallerPersons, CallerSessions} +} + +// EffectiveCallers are the callers the verb is open to: the list it names, or the default of its class. A verb whose +// callers are text, as the definitions before G wrote them, names no list and takes the default. +func (v Verb) EffectiveCallers() []string { + if v.Callers != nil && !v.Callers.IsText() { + return v.Callers.List + } + return DefaultCallers(v.Class) +} + // The classes of a verb (design 47 G §391). const ( ClassRead = "read" diff --git a/go/seatdef/seatdef_test.go b/go/seatdef/seatdef_test.go index abaa9720..206db19f 100644 --- a/go/seatdef/seatdef_test.go +++ b/go/seatdef/seatdef_test.go @@ -283,3 +283,22 @@ func TestAKeptEventsStreamIsNamedFromTheSeatAndTheEvent(t *testing.T) { t.Errorf("the stream is %s", got) } } + +// Design 47 G §391: a verb that names no callers takes its class's default, a verb that names them keeps them. +func TestAVerbWithoutCallersTakesItsClassDefault(t *testing.T) { + for _, c := range []struct { + verb Verb + want string + }{ + {Verb{}, "invokers persons sessions"}, + {Verb{Class: ClassRead}, "invokers persons sessions"}, + {Verb{Class: ClassAct}, "operator sessions"}, + {Verb{Class: ClassWarrant}, "operator"}, + {Verb{Class: ClassAct, Callers: &Callers{List: []string{"seat:mesh-broker"}}}, "seat:mesh-broker"}, + {Verb{Class: ClassWarrant, Callers: &Callers{Text: "a person"}}, "operator"}, + } { + if got := strings.Join(c.verb.EffectiveCallers(), " "); got != c.want { + t.Errorf("%+v is open to %q, want %q", c.verb, got, c.want) + } +} +} -- 2.54.0 From 57b1be1bd56cc44812fce78909bac8c1fbd75828 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 10:38:11 +0100 Subject: [PATCH 002/134] D-3: a kept event's stream is SEAT__, bounded by its keep object MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit keep is an object (max-age-days, max-mib, last-per-subject; 90 days and 16 MiB by default, 0 no age limit); the string form and the message-level bounds go; an event with a keep has no dot in its name. D-2 (step pipeline, 30 min default; on kind:image) was built in the reader release. Builds: design 47 G §260, §331, §396, §397, §203, §266, §324 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/seatdef/seatdef.go | 60 ++++++++++++++++++-------------------- go/seatdef/seatdef_test.go | 37 +++++++++++++++++------ 2 files changed, 57 insertions(+), 40 deletions(-) diff --git a/go/seatdef/seatdef.go b/go/seatdef/seatdef.go index d2838992..143657a5 100644 --- a/go/seatdef/seatdef.go +++ b/go/seatdef/seatdef.go @@ -244,29 +244,22 @@ type Message struct { // the grants part grants a node's account alone (ADR 0363 §2). Subject string `json:"subject,omitempty"` // Keep says how an event is kept when not as the family keeps the seat's others, in a stream of its own named - // KeptStream: KeepLastPerSubject, the last message of each subject and no age limit (design 47: the node-engine's - // running.., so the last build of every module on every node stays on the bus however long ago it - // was said, and a durable consumer starting from the last message per subject holds them all); or KeepStream, - // every message within MaxAgeDays and MaxMiB. + // KeptStream (design 47 G §396): its age, its cap and whether it keeps the last message of each subject only (the + // node-engine's running.., so the last build of every module on every node stays on the bus however + // long ago it was said). An event with a Keep has no dot in its name. Keep *Keep `json:"keep,omitempty"` // Callers says who may send an accepted kind, as a verb's (design 47 G §338). Callers *Callers `json:"callers,omitempty"` // Stream is the stream an accepted kind rides, StreamCheck or StreamDelivery; its subject is `.`, and // the same kind name may ride more than one stream (design 47 G §415). Stream string `json:"stream,omitempty"` - // MaxAgeDays and MaxMiB bound an event kept KeepStream: how long a message is kept and how much the stream holds - // (design 47 *What the gate owes the forge*: the stream `owed`, 7 days and at most 64 MiB). Given with KeepStream - // alone. - MaxAgeDays int `json:"max-age-days,omitempty"` - MaxMiB int `json:"max-mib,omitempty"` } -// The ways an event is kept in a stream of its own, KeptStream, instead of its seat's family. +// What an event's keep says where it says less (design 47 G §396, §331). const ( - // KeepLastPerSubject keeps the last message of each subject, with no age limit. - KeepLastPerSubject = "last-per-subject" - // KeepStream keeps every message for MaxAgeDays, at most MaxMiB in all. - KeepStream = "stream" + // KeepDefaultAgeDays and KeepDefaultMiB are an event's age and cap when its keep names none. + KeepDefaultAgeDays = 90 + KeepDefaultMiB = 16 ) // KeptStream is the stream an event with Keep is kept in: `SEAT__` (design 47 G §260, §396). @@ -275,14 +268,26 @@ func KeptStream(seat string, m Message) string { return KeptStreamPrefix + seat // KeptStreamPrefix begins the name of every kept event's stream. const KeptStreamPrefix = "SEAT_" -// legacyMode is the string form of Keep, empty for none or for the object form. -func (m Message) legacyMode() string { - if m.Keep == nil { - return "" +// KeptAgeDays is how long a kept event's stream keeps a message: the keep's age, 90 days when it names none, and 0 +// for no age limit. +func (m Message) KeptAgeDays() int { + if m.Keep == nil || m.Keep.MaxAgeDays == nil { + return KeepDefaultAgeDays } - return m.Keep.Mode + return *m.Keep.MaxAgeDays } +// KeptMiB is the most a kept event's stream holds: the keep's cap, 16 MiB when it names none. +func (m Message) KeptMiB() int { + if m.Keep == nil || m.Keep.MaxMiB == nil { + return KeepDefaultMiB + } + return *m.Keep.MaxMiB +} + +// KeptLastPerSubject says the stream keeps only the last message of each subject. +func (m Message) KeptLastPerSubject() bool { return m.Keep != nil && m.Keep.LastPerSubject } + // SubjectPattern is the event's subject after the seat's event prefix with its parts between angle brackets: the // name itself when the definition gives no subject. func (m Message) SubjectPattern() string { @@ -833,26 +838,19 @@ func checkMessages(field string, ms []Message, whole bool) Problems { case m.Stream != "" && m.Stream != StreamCheck && m.Stream != StreamDelivery: ps = append(ps, Problem{where + ".stream", fmt.Sprintf("%q is not %s or %s", m.Stream, StreamCheck, StreamDelivery)}) } - mode := m.legacyMode() switch { case m.Keep != nil && field != "events": ps = append(ps, Problem{where + ".keep", "only an event is kept"}) - case m.Keep != nil && m.Keep.Object: + case m.Keep != nil && !m.Keep.Object: + ps = append(ps, Problem{where + ".keep", fmt.Sprintf("%q is the old form: keep is an object of max-age-days, max-mib and last-per-subject", m.Keep.Mode)}) + case m.Keep != nil: ps = append(ps, checkKeepObject(where+".keep", *m.Keep)...) - if m.MaxAgeDays != 0 || m.MaxMiB != 0 { - ps = append(ps, Problem{where + ".max-age-days", "an event whose keep is an object gives its max-age-days and max-mib there"}) + if strings.Contains(m.Name, ".") { + ps = append(ps, Problem{where + ".name", "an event with a keep has no dot in its name: its stream is named from it"}) } if m.Keep.LastPerSubject && len(m.Parts()) == 0 && !m.ByCaller { ps = append(ps, Problem{where + ".keep", "an event of one subject kept last per subject is one message: give it a subject with parts"}) } - case mode != "" && mode != KeepLastPerSubject && mode != KeepStream: - ps = append(ps, Problem{where + ".keep", fmt.Sprintf("%q is not %s or %s", mode, KeepLastPerSubject, KeepStream)}) - case mode == KeepLastPerSubject && len(m.Parts()) == 0 && !m.ByCaller: - ps = append(ps, Problem{where + ".keep", "an event of one subject kept last per subject is one message: give it a subject with parts"}) - case mode == KeepStream && (m.MaxAgeDays <= 0 || m.MaxMiB <= 0): - ps = append(ps, Problem{where + ".keep", "an event kept in a stream of its own gives its max-age-days and max-mib"}) - case mode != KeepStream && (m.MaxAgeDays != 0 || m.MaxMiB != 0): - ps = append(ps, Problem{where + ".max-age-days", "only an event kept as a stream of its own gives max-age-days and max-mib"}) } if m.Callers != nil && field != "accepts" { ps = append(ps, Problem{where + ".callers", "only an accepted kind carries callers"}) diff --git a/go/seatdef/seatdef_test.go b/go/seatdef/seatdef_test.go index 206db19f..e68e2969 100644 --- a/go/seatdef/seatdef_test.go +++ b/go/seatdef/seatdef_test.go @@ -2,6 +2,7 @@ package seatdef import ( "errors" + "fmt" "strings" "testing" ) @@ -64,17 +65,18 @@ func TestADefinitionIsReadStrictlyNamingWhatIsWrong(t *testing.T) { {"a subject on an event named by its caller", strings.Replace(whole, `"name":"created"`, `"name":"owed","by-caller":true,"subject":"owed."`, 1), "carries no subject of its own"}, {"an event kept last per subject", strings.Replace(whole, `"name":"created"`, - `"name":"running","subject":"running..","keep":"last-per-subject"`, 1), ""}, - {"an event kept otherwise", strings.Replace(whole, `"name":"created"`, - `"name":"running","subject":"running.","keep":"forever"`, 1), `events[0].keep: "forever" is not last-per-subject or stream`}, + `"name":"running","subject":"running..","keep":{"last-per-subject":true,"max-age-days":0}`, 1), ""}, + {"an event kept in the old form", strings.Replace(whole, `"name":"created"`, + `"name":"running","subject":"running.","keep":"last-per-subject"`, 1), `events[0].keep: "last-per-subject" is the old form`}, {"an event kept in a stream of its own, bounded", strings.Replace(whole, `"name":"created"`, - `"name":"owed","subject":"owed..","keep":"stream","max-age-days":7,"max-mib":64`, 1), ""}, - {"an event kept in a stream of its own, unbounded", strings.Replace(whole, `"name":"created"`, - `"name":"owed","subject":"owed..","keep":"stream","max-age-days":7`, 1), "gives its max-age-days and max-mib"}, - {"a bound on an event kept by its family", strings.Replace(whole, `"name":"created"`, - `"name":"owed","max-mib":64`, 1), "events[0].max-age-days: only an event kept as a stream"}, + `"name":"owed","subject":"owed..","keep":{"max-age-days":7,"max-mib":64}`, 1), ""}, + {"an event kept with a dot in its name", strings.Replace(whole, `"name":"created"`, + `"name":"pull.updated","keep":{}`, 1), "events[0].name: an event with a keep has no dot"}, + {"an event kept by its family may have a dot", strings.Replace(whole, `"name":"created"`, `"name":"pull.updated"`, 1), ""}, + {"a bound beside no keep", strings.Replace(whole, `"name":"created"`, + `"name":"owed","max-mib":64`, 1), "max-mib: a field the format does not know"}, {"an event of one subject kept last per subject", strings.Replace(whole, `"name":"created"`, - `"name":"running","keep":"last-per-subject"`, 1), "give it a subject with parts"}, + `"name":"running","keep":{"last-per-subject":true}`, 1), "give it a subject with parts"}, {"an accept with dotted words", strings.Replace(whole, `"events":`, `"accepts":[{"name":"run.now","body":{}}],"events":`, 1), "accepts[0].name"}, {"a verb twice", strings.Replace(listGrain, `]}`, `,{"name":"add"}]}`, 1), `"add" twice`}, @@ -300,5 +302,22 @@ func TestAVerbWithoutCallersTakesItsClassDefault(t *testing.T) { if got := strings.Join(c.verb.EffectiveCallers(), " "); got != c.want { t.Errorf("%+v is open to %q, want %q", c.verb, got, c.want) } + } } + +// Design 47 G §260, §396, §331: a kept event's stream is named from the seat and the event, 90 days and 16 MiB unless +// the keep says otherwise, and 0 is no age limit. +func TestAKeptEventsStreamIsNamedAndBoundedByItsKeep(t *testing.T) { + d, err := Parse([]byte(`{"format":1,"seat":"mesh-delivery","scope":"mesh","grain":"seat-list","domain":"Delivery","decision":"x", + "events":[{"name":"owed","keep":{"max-age-days":7,"max-mib":64}},{"name":"toolchain-agreed","keep":{}}, + {"name":"versions-filled","keep":{"max-age-days":0,"max-mib":1}}]}`)) + if err != nil { + t.Fatal(err) + } + for i, want := range []string{"SEAT_mesh-delivery_owed 7 64", "SEAT_mesh-delivery_toolchain-agreed 90 16", "SEAT_mesh-delivery_versions-filled 0 1"} { + m := d.Events[i] + if got := fmt.Sprint(KeptStream(d.Seat, m), " ", m.KeptAgeDays(), " ", m.KeptMiB()); got != want { + t.Errorf("%s: %s, want %s", m.Name, got, want) + } + } } -- 2.54.0 From b81d338e58b1a26af54365061de36d97d99f0968 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 10:38:52 +0100 Subject: [PATCH 003/134] O-4: the call that reads a sent object MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §347, §407 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/buscalls/buscalls.go | 13 +++++++++++++ go/packages_list_test.go | 6 ++++++ 2 files changed, 19 insertions(+) diff --git a/go/buscalls/buscalls.go b/go/buscalls/buscalls.go index 44e007d4..b39dada2 100644 --- a/go/buscalls/buscalls.go +++ b/go/buscalls/buscalls.go @@ -28,6 +28,10 @@ const ( ObjectsDelete = "mesh/objects.delete" ObjectsInfo = "mesh/objects.info" ObjectsWatch = "mesh/objects.watch" + // ObjectsSentGet reads an object of a sender's store of sent objects by the reference a reply or an event carried + // (design 47 G §347, §407), served under the standing grant the grants part derives from the manifests; the answer is + // ObjectData. + ObjectsSentGet = "mesh/objects.sent-get" StreamRead = "mesh/stream.read" StreamFill = "mesh/stream.fill" @@ -103,6 +107,15 @@ type ObjectInfo struct { Deleted bool `json:"deleted,omitempty"` } +// ObjectsSentGetRequest names one sent object by the reference that carried it (stateobjects.Ref's fields, which this +// package does not import): the sender's store, the object's name, and its size and digest, which the reader checks. +type ObjectsSentGetRequest struct { + Store string `json:"store"` + Object string `json:"object"` + Size int64 `json:"size"` + Digest string `json:"sha256"` +} + // ObjectData is get's answer: the object's info and its bytes. type ObjectData struct { Info ObjectInfo `json:"info"` diff --git a/go/packages_list_test.go b/go/packages_list_test.go index a861aee3..35dfb78d 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -311,6 +311,12 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) { t.Errorf("%s: unwrapped %s, %q, %v; want %s, %q, %v", c.name, answer, said, ok, c.answer, c.said, c.ok) } } + // design 47 G §347: a sent object is asked by its reference's fields. + sentAsk := buscalls.ObjectsSentGetRequest{Store: "m_sent-objects", Object: "o", Size: 3 << 20, Digest: strings.Repeat("b", 64)} + if raw, err := json.Marshal(sentAsk); err != nil || buscalls.ObjectsSentGet != "mesh/objects.sent-get" || + string(raw) != `{"store":"m_sent-objects","object":"o","size":3145728,"sha256":"`+strings.Repeat("b", 64)+`"}` { + t.Errorf("the sent-get request is %s, %v", raw, err) + } // 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 -- 2.54.0 From 13a6891bc7a7f3ac49adbd770df577c744b39388 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 16:33:05 +0100 Subject: [PATCH 004/134] O-1: C-S1: one rule for the bus names of an entry in catalogue, the object store beside a state and the sent store (milestone 47/G, issue 1966, ADR 0666 section 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §58, §59 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/catalogue/busnames.go | 88 +++++++++++++++++++++++++++++++++++ go/catalogue/busnames_test.go | 38 +++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 go/catalogue/busnames.go create mode 100644 go/catalogue/busnames_test.go diff --git a/go/catalogue/busnames.go b/go/catalogue/busnames.go new file mode 100644 index 00000000..fc1cf052 --- /dev/null +++ b/go/catalogue/busnames.go @@ -0,0 +1,88 @@ +package catalogue + +import ( + "regexp" + "slices" + "sort" + "strings" +) + +// The names of what a module keeps on the bus, as one rule (design 25 § 12.1; novox/hq ADR 0666 §16): the bus module +// creates them from a module's entries and the grants part grants them, and both take the names from here. + +// BusNameSeparator joins a module's name and an entry's local name into the name the mesh gives the entry on the bus. +const BusNameSeparator = "_" + +// SentObjectsLocal is the local name of a module's store of sent objects (the bodies of a reply or an event over the +// per-message limit): the bus module plans `_sent-objects` for every module, and no entry may take the name. +const SentObjectsLocal = "sent-objects" + +// ObjectsSuffix ends the local name of the object store beside a state: `-objects`. +const ObjectsSuffix = "-objects" + +// safeToken is a name that is one plain subject token. +var safeToken = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) + +// BusName is `_`, the name the mesh gives an entry on the bus. +func BusName(module, local string) string { return module + BusNameSeparator + local } + +// StateObjectsName is the local name of the object store that holds the large values of the state `state`. +func StateObjectsName(state string) string { return state + ObjectsSuffix } + +// SentObjectsName is the bus name of a module's store of sent objects. +func SentObjectsName(module string) string { return BusName(module, SentObjectsLocal) } + +// BusStateStreams are the streams one bus-state entry of module is kept in on the bus, by the mesh's names +// `_`: a stream's own; a key-value bucket's KV_ stream with the OBJ_ stream of its object store +// `-objects`; an object store's OBJ_ stream; a ledger's history stream, its snapshot stream, and its state and +// index buckets' KV_ streams. A name that is not one plain subject token names nothing. +func BusStateStreams(module string, e BusStateEntry) []string { + base := BusName(module, e.Name) + if !safeToken.MatchString(base) || strings.Contains(e.Name, BusNameSeparator) || strings.Contains(module, BusNameSeparator) { + return nil + } + var out []string + switch e.Storage { + case StorageKeyValue: + out = []string{"KV_" + base} + if e.Template == TemplateState { + out = append(out, "OBJ_"+BusName(module, StateObjectsName(e.Name))) + } + case StorageObjectStore: + out = []string{"OBJ_" + base} + case StorageLedger: + out = []string{base, base + "_snapshot", "KV_" + base + "_state"} + for _, t := range e.Types { + for _, ix := range t.Indexes { + if safeToken.MatchString(ix.Name) { + out = append(out, "KV_"+base+"_index-"+ix.Name) + } + } + } + default: + out = []string{base} + } + sort.Strings(out) + return slices.Compact(out) +} + +// ObjectNameClashes are the local names of a module's bus-state entries that meet a name the mesh makes itself: the +// name of a state's object store (`-objects` beside the state ``) and the store of sent objects. The +// indexer's judge refuses each. +func ObjectNameClashes(entries []BusStateEntry) []string { + states := map[string]bool{} + for _, e := range entries { + if e.Storage == StorageKeyValue && e.Template == TemplateState { + states[e.Name] = true + } + } + var out []string + for _, e := range entries { + stem, ok := strings.CutSuffix(e.Name, ObjectsSuffix) + if e.Name == SentObjectsLocal || (ok && states[stem]) { + out = append(out, e.Name) + } + } + sort.Strings(out) + return slices.Compact(out) +} diff --git a/go/catalogue/busnames_test.go b/go/catalogue/busnames_test.go new file mode 100644 index 00000000..087b6484 --- /dev/null +++ b/go/catalogue/busnames_test.go @@ -0,0 +1,38 @@ +package catalogue + +import ( + "slices" + "testing" +) + +// ADR 0666 §16: the bus names of an entry are one rule. A state entry answers its bucket's stream and the stream of +// its object store; a local name that meets a name the mesh makes is a clash for the judge. +func TestTheBusNamesOfAnEntryAreOneRule(t *testing.T) { + state := BusStateEntry{Name: "views", Storage: StorageKeyValue, Template: TemplateState} + rows := []struct { + name string + entry BusStateEntry + streams []string + }{ + {"a state: its bucket and its object store", state, []string{"KV_m_views", "OBJ_m_views-objects"}}, + {"a key-value that is not a state template: its bucket only", BusStateEntry{Name: "x", Storage: StorageKeyValue}, []string{"KV_m_x"}}, + {"an object store", BusStateEntry{Name: "blobs", Storage: StorageObjectStore, Template: TemplateObjects}, []string{"OBJ_m_blobs"}}, + {"a stream", BusStateEntry{Name: "log", Storage: StorageStream, Template: TemplateLog}, []string{"m_log"}}, + {"a ledger", BusStateEntry{Name: "l", Storage: StorageLedger, Template: TemplateLedger}, []string{"KV_m_l_state", "m_l", "m_l_snapshot"}}, + {"a name that is not one token", BusStateEntry{Name: "a.b", Storage: StorageStream}, nil}, + {"a local name holding the separator", BusStateEntry{Name: "a_b", Storage: StorageStream}, nil}, + } + for _, r := range rows { + if got := BusStateStreams("m", r.entry); !slices.Equal(got, r.streams) { + t.Errorf("%s: streams %v, want %v", r.name, got, r.streams) + } + } + if got := SentObjectsName("m"); got != "m_sent-objects" { + t.Errorf("sent store %q", got) + } + clash := []BusStateEntry{state, {Name: "views-objects", Storage: StorageObjectStore}, {Name: "other-objects", Storage: StorageObjectStore}, + {Name: SentObjectsLocal, Storage: StorageObjectStore}} + if got := ObjectNameClashes(clash); !slices.Equal(got, []string{"sent-objects", "views-objects"}) { + t.Errorf("clashes %v", got) + } +} -- 2.54.0 From b1391867a882333caa3e9333e9cf481c415304da Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 16:34:14 +0100 Subject: [PATCH 005/134] O-2: C-S2: stateobjects.Ref, a body exactly {mesh-object: Ref} is a reference and anything else a value (milestone 47/G, issue 1966, ADR 0666 section 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §57 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/packages_list_test.go | 2 +- go/stateobjects/ref.go | 71 +++++++++++++++++++++++++++++++++++++ go/stateobjects/ref_test.go | 31 ++++++++++++++++ 3 files changed, 103 insertions(+), 1 deletion(-) create mode 100644 go/stateobjects/ref.go create mode 100644 go/stateobjects/ref_test.go diff --git a/go/packages_list_test.go b/go/packages_list_test.go index 35dfb78d..109974d4 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -26,7 +26,7 @@ var meshSDKPackages = []string{ ".", "announce", "announcetest", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "declaration", "forgekey", "health", "internal/beside", "keys", "kvlimit", "membership", "motion", "objectlink", "outward", "pernode", "process", - "sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "testbus", "views", "views/viewstest", + "sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "stateobjects", "testbus", "views", "views/viewstest", } // clientSeats are the seats whose logic is their client's in mesh-contracts (novox/hq ADR 0627 §1, §4): mesh-sdk names diff --git a/go/stateobjects/ref.go b/go/stateobjects/ref.go new file mode 100644 index 00000000..1c76fc79 --- /dev/null +++ b/go/stateobjects/ref.go @@ -0,0 +1,71 @@ +// 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) diff --git a/go/stateobjects/ref_test.go b/go/stateobjects/ref_test.go new file mode 100644 index 00000000..1fe64ae8 --- /dev/null +++ b/go/stateobjects/ref_test.go @@ -0,0 +1,31 @@ +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 TestABodyIsAReferenceOnlyWhenItIsExactlyOne(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) + } + } +} -- 2.54.0 From 786397025151400a61363b509456e1d6ab347e17 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 16:37:28 +0100 Subject: [PATCH 006/134] O-2: C-S3, C-S5: stateobjects.Values over a bundle's runtime, values of any size, a replace removes the object it replaced, a deleted key leaves its object listed key-gone (milestone 47/G, issue 1966, ADR 0666 section 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runtime's channel carries 16 MiB a message, so a value over the 8 MiB a put takes is split over objects (Ref.Parts). Builds: design 47 G §57, §138 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/stateobjects/bundle.go | 81 +++++++++ go/stateobjects/fakes_test.go | 228 +++++++++++++++++++++++++ go/stateobjects/ref.go | 5 +- go/stateobjects/values.go | 303 +++++++++++++++++++++++++++++++++ go/stateobjects/values_test.go | 188 ++++++++++++++++++++ 5 files changed, 804 insertions(+), 1 deletion(-) create mode 100644 go/stateobjects/bundle.go create mode 100644 go/stateobjects/fakes_test.go create mode 100644 go/stateobjects/values.go create mode 100644 go/stateobjects/values_test.go diff --git a/go/stateobjects/bundle.go b/go/stateobjects/bundle.go new file mode 100644 index 00000000..930e554b --- /dev/null +++ b/go/stateobjects/bundle.go @@ -0,0 +1,81 @@ +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 +// `-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 } + +// BundleObjectMost is the largest object a bundle puts in one call: the runtime's channel carries a message of 16 MiB, and +// the bytes travel in it as base64. +const BundleObjectMost = 8 << 20 + +// MostBytes is BundleObjectMost: a larger value is split over objects. +func (BundleObjects) MostBytes() int { return BundleObjectMost } + +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 +} diff --git a/go/stateobjects/fakes_test.go b/go/stateobjects/fakes_test.go new file mode 100644 index 00000000..b0a6a534 --- /dev/null +++ b/go/stateobjects/fakes_test.go @@ -0,0 +1,228 @@ +package stateobjects + +import ( + "bufio" + "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 + most int +} + +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 + 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))}) + } + 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"` + } + _ = 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 buscalls.ObjectsPut: + return buscalls.ObjectInfo{Name: p.Name, Size: uint64(len(p.Data))}, m.Put(p.Name, p.Data) + case buscalls.ObjectsGet: + d, err := m.Object(p.Name) + return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Name}, Data: d}, 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) +} diff --git a/go/stateobjects/ref.go b/go/stateobjects/ref.go index 1c76fc79..008f4e55 100644 --- a/go/stateobjects/ref.go +++ b/go/stateobjects/ref.go @@ -27,6 +27,9 @@ type Ref struct { Object string `json:"object"` // Size is the object's size in bytes. Size int64 `json:"size"` + // Parts is how many objects the bytes are split over, named `.` from 0, when the store takes no object as + // large as the value (a bundle's runtime carries a message of 16 MiB); zero is one object named Object. + Parts int `json:"parts,omitempty"` // 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. @@ -34,7 +37,7 @@ type Ref struct { } func (r Ref) valid() bool { - if r.Store == "" || r.Object == "" || r.Size < 0 || r.Form != FormVersion || len(r.Digest) != 64 { + if r.Store == "" || r.Object == "" || r.Size < 0 || r.Parts < 0 || r.Form != FormVersion || len(r.Digest) != 64 { return false } _, err := hex.DecodeString(r.Digest) diff --git a/go/stateobjects/values.go b/go/stateobjects/values.go new file mode 100644 index 00000000..4c36cb90 --- /dev/null +++ b/go/stateobjects/values.go @@ -0,0 +1,303 @@ +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 split over objects. + 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, `.`: 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 `.` with its reference +// in the key. A replace removes the object it replaced in the same act; 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, "" + var newRef Ref + 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 !(hadRef && oldRef.Object == object) { + parts, err := putObject(v.Objects, object, data) + if err != nil { + return fmt.Errorf("the object for %s: %w", key, err) + } + ref.Parts = parts + } else { + ref.Parts = oldRef.Parts + } + body = Wrap(ref) + newRef = 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) { + _ = deleteObject(v.Objects, newRef) + } + return err + } + if hadRef && oldRef.Object != object { + if err := deleteObject(v.Objects, oldRef); err != nil { + return fmt.Errorf("%s is written; the object it replaced, %s, could not be removed: %w", key, oldRef.Object, 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 = getObject(v.Objects, ref); 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 { + for _, name := range ref.names() { + live[name] = 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 +} + +// names are the objects a reference stands for. +func (r Ref) names() []string { + if r.Parts <= 1 { + return []string{r.Object} + } + out := make([]string, r.Parts) + for i := range out { + out[i] = fmt.Sprintf("%s.%d", r.Object, i) + } + return out +} + +// keyOf is the key an object `.` or `..` was written for. +func keyOf(object string) string { + isDigest := func(s string) bool { _, err := hex.DecodeString(s); return len(s) == 64 && err == nil } + parts := strings.Split(object, ".") + if n := len(parts); n >= 3 && isDigest(parts[n-2]) { + return strings.Join(parts[:n-2], ".") + } + if n := len(parts); n >= 2 && isDigest(parts[n-1]) { + return strings.Join(parts[:n-1], ".") + } + return object +} + +// putObject stores data as one object, or over `.` when the store takes none so large; it answers the parts +// (zero for one object) and removes what it stored when a part fails. +func putObject(o Objects, name string, data []byte) (int, error) { + most := o.MostBytes() + if most <= 0 || len(data) <= most { + return 0, o.Put(name, data) + } + ref := Ref{Object: name, Parts: (len(data) + most - 1) / most} + for i, part := range ref.names() { + from, to := i*most, min((i+1)*most, len(data)) + if err := o.Put(part, data[from:to]); err != nil { + _ = deleteObject(o, Ref{Object: name, Parts: i}) + return 0, err + } + } + return ref.Parts, nil +} + +func getObject(o Objects, ref Ref) ([]byte, error) { + var out []byte + for _, name := range ref.names() { + d, err := o.Get(name) + if err != nil { + return nil, err + } + out = append(out, d...) + } + return out, nil +} + +func deleteObject(o Objects, ref Ref) error { + var first error + for _, name := range ref.names() { + if err := o.Delete(name); err != nil && first == nil { + first = err + } + } + return first +} diff --git a/go/stateobjects/values_test.go b/go/stateobjects/values_test.go new file mode 100644 index 00000000..16f93329 --- /dev/null +++ b/go/stateobjects/values_test.go @@ -0,0 +1,188 @@ +package stateobjects + +import ( + "bytes" + "encoding/json" + "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 one object, a delete leaves one listed key-gone (ADR 0666 §16, C-Q1). +func valueRows(t *testing.T, v Values) { + t.Helper() + limit := kvlimit.ModuleStateValueBytes + for _, times := range []int{1, 10, 100} { + 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 removes the object it replaced; the same value again leaves its object + before, _ := v.Objects.List() + if err := v.Put("one", big(2*limit)); err != nil { // 10x key + 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) { + t.Errorf("a replace left %d objects, %d before it", len(after), len(before)) + return + } + // a replace by a small value removes the object too + if err := v.Put("one", "small now"); err != nil { + t.Error(err) + return + } + if now, _ := v.Objects.List(); len(now) != len(before)-1 { + t.Errorf("replacing a large value by a small one left %d objects, want %d", len(now), len(before)-1) + 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) + } + } + if len(gone) != 1 || gone[0] != "ten" { + t.Errorf("the objects whose key is gone: %v, want ten's", gone) + return + } + dropped, err := v.DropKeyGone() + if err != nil || len(dropped) != 1 { + t.Errorf("dropped %v, %v", dropped, err) + return + } + if left, _ := v.Objects.List(); len(left) != len(kept)-1 { + 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 TestValuesOfAnySizeThroughTheRuntime(t *testing.T) { + m := newMemory() + out := runtimeOf(t, m, func() (any, error) { + v := ForState("views") + valueRows(t, v) + 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 TestAWriteLostLeavesNoObjectBehind(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 gets it split over objects, read back whole, replaced and dropped as +// one (a bundle's runtime carries a message of 16 MiB). +func TestAValueLargerThanTheStoreTakesIsSplitOverObjects(t *testing.T) { + m := newMemory() + m.most = 100_000 + v := m.values() + want := big(5*kvlimit.ModuleStateValueBytes + 10) + if err := v.Put("k", want); err != nil { + t.Fatal(err) + } + if l, _ := (memObjects{m}).List(); len(l) < 13 { + t.Fatalf("%d objects, want the value split over at least 13", len(l)) + } + var got map[string]string + if ok, err := v.Get("k", &got); err != nil || !ok || got["text"] != want["text"] { + t.Fatalf("read back %d, %v, %v", len(got["text"]), ok, err) + } + if err := v.Delete("k"); err != nil { + t.Fatal(err) + } + kept, _ := v.Kept() + for _, k := range kept { + if !k.KeyGone || k.Key != "k" { + t.Fatalf("a part %+v is not listed as key-gone for k", k) + } + } + if dropped, err := v.DropKeyGone(); err != nil || len(dropped) != len(kept) { + t.Fatalf("dropped %v of %d, %v", dropped, len(kept), err) + } +} -- 2.54.0 From cb86079051cd988fb17f2e0db9e37be796a72dd9 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 16:37:59 +0100 Subject: [PATCH 007/134] O-3: C-S4: stateobjects.OverBus, the same values of any size through a process's own JetStream, tested against the bus (milestone 47/G, issue 1966, ADR 0666 section 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §58, §407 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/stateobjects/process.go | 126 +++++++++++++++++++++ go/stateobjects/values_integration_test.go | 39 +++++++ 2 files changed, 165 insertions(+) create mode 100644 go/stateobjects/process.go create mode 100644 go/stateobjects/values_integration_test.go diff --git a/go/stateobjects/process.go b/go/stateobjects/process.go new file mode 100644 index 00000000..d799c27a --- /dev/null +++ b/go/stateobjects/process.go @@ -0,0 +1,126 @@ +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 `-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 +} diff --git a/go/stateobjects/values_integration_test.go b/go/stateobjects/values_integration_test.go new file mode 100644 index 00000000..be0dad25 --- /dev/null +++ b/go/stateobjects/values_integration_test.go @@ -0,0 +1,39 @@ +//go:build integration + +package stateobjects + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// 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) + } + t.Run("values of 1x, 10x and 100x the limit", func(t *testing.T) { valueRows(t, OverBus(kv, store, "m_views-objects")) }) +} -- 2.54.0 From 71f8e548aa609ebe42482f033d54389bbff66851 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 16:38:39 +0100 Subject: [PATCH 008/134] O-5: C-S6: stateobjects.Sender and Receive, a reply or an event over the message limit sent as the reference of an object in the sender's store of sent objects (milestone 47/G, issue 1966, ADR 0666 section 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §59, §407 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/stateobjects/fakes_test.go | 4 + go/stateobjects/send.go | 93 ++++++++++++++++++++++ go/stateobjects/send_test.go | 68 ++++++++++++++++ go/stateobjects/values_integration_test.go | 20 +++++ 4 files changed, 185 insertions(+) create mode 100644 go/stateobjects/send.go create mode 100644 go/stateobjects/send_test.go diff --git a/go/stateobjects/fakes_test.go b/go/stateobjects/fakes_test.go index b0a6a534..60c454da 100644 --- a/go/stateobjects/fakes_test.go +++ b/go/stateobjects/fakes_test.go @@ -192,6 +192,7 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) { Revision uint64 `json:"revision"` Name string `json:"name"` Data []byte `json:"data"` + Object string `json:"object"` } _ = json.Unmarshal(raw, &p) switch method { @@ -214,6 +215,9 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) { case buscalls.ObjectsGet: d, err := m.Object(p.Name) return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Name}, Data: d}, err + case buscalls.ObjectsSentGet: + d, err := m.Object(p.Object) + return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Object}, Data: d}, err case buscalls.ObjectsDelete: return buscalls.Done{}, memObjects{m}.Delete(p.Name) case buscalls.ObjectsList: diff --git a/go/stateobjects/send.go b/go/stateobjects/send.go new file mode 100644 index 00000000..6420a2ed --- /dev/null +++ b/go/stateobjects/send.go @@ -0,0 +1,93 @@ +package stateobjects + +import ( + "encoding/json" + "fmt" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// 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 `_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 +} + +// Send is what to send for body: the body itself when it fits, else the reference of the object it was put in as +// `