From 48706563c36ec0ab40960442526044532d3135d5 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 21:54:31 +0100 Subject: [PATCH] G C-2, C-3: a value larger than one chunk is put and got by session through the tool runner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit buscalls.ChunkBytes (512 KiB), mesh/objects.put-chunk and get-chunk with the store named in each chunk, the digest and size of a get checked, the capability word put-sessions required, and a bundle's refusal of a value over 8 MiB gone. Builds: design 47 G §518, §345, §490, §620, A-21 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/buscalls/buscalls.go | 51 ++++++++++++++ go/busstate/busstate.go | 27 +++++++- go/busstate/session.go | 110 +++++++++++++++++++++++++++++ go/busstate/session_test.go | 123 +++++++++++++++++++++++++++++++++ go/stateobjects/bundle.go | 9 +-- go/stateobjects/fakes_test.go | 48 +++++++++++-- go/stateobjects/send.go | 8 +-- go/stateobjects/send_test.go | 7 +- go/stateobjects/values_test.go | 2 +- 9 files changed, 362 insertions(+), 23 deletions(-) create mode 100644 go/busstate/session.go create mode 100644 go/busstate/session_test.go diff --git a/go/buscalls/buscalls.go b/go/buscalls/buscalls.go index 00d54326..2d7ef190 100644 --- a/go/buscalls/buscalls.go +++ b/go/buscalls/buscalls.go @@ -32,6 +32,11 @@ const ( // (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" + // ObjectsPutChunk and ObjectsGetChunk are the put session's and the get session's calls (design 47 G §518, §345, §490), + // under the capability word put-sessions (stdio.CapPutSessions): a value larger than ChunkBytes is put as chunks in order, + // each answered before the next, and a get that answers ObjectsChunked is read chunk by chunk. + ObjectsPutChunk = "mesh/objects.put-chunk" + ObjectsGetChunk = "mesh/objects.get-chunk" StreamRead = "mesh/stream.read" StreamFill = "mesh/stream.fill" @@ -94,6 +99,52 @@ type ObjectsPutRequest struct { Description string `json:"description,omitempty"` } +// ChunkBytes is the size of one chunk of a put or get session, and so the largest value one call puts or gets (design 47 G +// §518, A-21): it fits under the bus's default maximum payload, whatever the bus sets it to. +const ChunkBytes = 512 << 10 + +// ObjectsPutChunkRequest is one chunk of a put session (design 47 G §518, A-21): the object `name` in the store `objects`, +// `call` the bundle's id for this put, `seq` from 1 and one more each chunk, `last` true on the last chunk alone, and the +// chunk's bytes. +type ObjectsPutChunkRequest struct { + Objects string `json:"objects"` + Name string `json:"name"` + Call string `json:"call"` + Seq uint32 `json:"seq"` + Last bool `json:"last,omitempty"` + Data []byte `json:"data"` +} + +// ObjectsChunkTaken is the answer to a chunk that is not the last: the sequence taken, or the reason it was not. +type ObjectsChunkTaken struct { + Seq uint32 `json:"seq"` + Error string `json:"error,omitempty"` +} + +// ObjectsChunked is the answer of a get of a value over one chunk (design 47 G §490, §518, §620): the count of chunks (or +// true, for a sent object), the whole size, its `sha256:` digest and the session's `call`. Its Chunked is absent from the +// answer of a value read in one call. +type ObjectsChunked struct { + Chunked json.RawMessage `json:"chunked"` + Size uint64 `json:"size"` + Digest string `json:"digest"` + Call string `json:"call"` +} + +// ObjectsGetChunkRequest asks one chunk of a get session by its call and sequence. +type ObjectsGetChunkRequest struct { + Call string `json:"call"` + Seq uint32 `json:"seq"` +} + +// ObjectsChunk is a get session's chunk: the sequence, whether it is the last, its bytes, or the reason it is none. +type ObjectsChunk struct { + Seq uint32 `json:"seq"` + Last bool `json:"last"` + Data []byte `json:"data"` + Error string `json:"error,omitempty"` +} + // ObjectsNameRequest names one object of an object store: get, delete. type ObjectsNameRequest struct { Objects string `json:"objects"` diff --git a/go/busstate/busstate.go b/go/busstate/busstate.go index d105dc10..f81adfba 100644 --- a/go/busstate/busstate.go +++ b/go/busstate/busstate.go @@ -31,14 +31,39 @@ func call[T any](method string, params any) (T, error) { } // Put stores one object whole under its name, replacing one of that name, and answers it as the store holds it. +// +// A value larger than buscalls.ChunkBytes is put as a session of chunks (design 47 G §518, §345): every object is one +// object, never split over several. func (o Objects) Put(name string, data []byte, description string) (buscalls.ObjectInfo, error) { + if len(data) > buscalls.ChunkBytes { + return o.putSession(name, data) + } return call[buscalls.ObjectInfo](buscalls.ObjectsPut, buscalls.ObjectsPutRequest{Objects: o.local, Name: name, Data: data, Description: description}) } // Get is one object, its info and its bytes. +// +// A value the runtime answers as a session (`chunked`) is read chunk by chunk and checked against its digest (design 47 G §490). func (o Objects) Get(name string) (buscalls.ObjectData, error) { - return call[buscalls.ObjectData](buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: o.local, Name: name}) + raw, err := stdio.BusState(buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: o.local, Name: name}) + if err != nil { + return buscalls.ObjectData{}, err + } + return ReadObject(name, raw) +} + +// ReadObject is the object a runtime's answer to a get (or to a sent-get, design 47 G §620) holds: the value itself, or the +// session of chunks it opened, read whole. +func ReadObject(name string, raw json.RawMessage) (buscalls.ObjectData, error) { + if c, ok := chunkedAnswer(raw); ok { + return getSession(name, c) + } + var out buscalls.ObjectData + if err := json.Unmarshal(raw, &out); err != nil { + return out, fmt.Errorf("the runtime's answer for %s: %w", name, err) + } + return out, nil } // List is every object the store holds. diff --git a/go/busstate/session.go b/go/busstate/session.go new file mode 100644 index 00000000..fbeee45d --- /dev/null +++ b/go/busstate/session.go @@ -0,0 +1,110 @@ +package busstate + +import ( + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// The put and get sessions of a value larger than one chunk (design 47 G §518, §345, §490, A-21): a bundle sends and asks the +// chunks as calls on its channel to the tool runner, which holds the session; nothing of it is on the bus. The capability +// word put-sessions names the two calls, so a tool runner that does not state it refuses the session at once. + +// callID is the bundle's id for one session: random, so two sessions never share one. +func callID() string { + var b [12]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} + +// put sends data as a put session: chunks of buscalls.ChunkBytes in order, each answered before the next is sent, the +// answer to the last being the put's. +func (o Objects) putSession(name string, data []byte) (buscalls.ObjectInfo, error) { + if err := stdio.RequireCapability(stdio.CapPutSessions); err != nil { + return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, larger than one call carries: %w", name, err) + } + id := callID() + for seq, at := uint32(1), 0; ; seq++ { + end := min(at+buscalls.ChunkBytes, len(data)) + last := end == len(data) + raw, err := stdio.BusState(buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: o.local, Name: name, Call: id, + Seq: seq, Last: last, Data: data[at:end]}) + if err != nil { + return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, chunk %d: %w", name, seq, err) + } + var taken buscalls.ObjectsChunkTaken + if err := json.Unmarshal(raw, &taken); err != nil { + return buscalls.ObjectInfo{}, fmt.Errorf("the runtime's answer to a chunk of %s: %w", name, err) + } + if taken.Error != "" { + return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, chunk %d: %s", name, seq, taken.Error) + } + if last { + var info buscalls.ObjectInfo + if err := json.Unmarshal(raw, &info); err != nil { + return info, fmt.Errorf("the runtime's answer to the put of %s: %w", name, err) + } + return info, nil + } + if taken.Seq != seq { + return buscalls.ObjectInfo{}, fmt.Errorf("putting %s: chunk %d was answered as chunk %d", name, seq, taken.Seq) + } + at = end + } +} + +// getSession reads the chunks of the session the runtime opened for name, and checks the whole against the digest and the +// size it answered. +func getSession(name string, c buscalls.ObjectsChunked) (buscalls.ObjectData, error) { + if err := stdio.RequireCapability(stdio.CapPutSessions); err != nil { + return buscalls.ObjectData{}, fmt.Errorf("getting %s, larger than one call carries: %w", name, err) + } + var whole []byte + for seq := uint32(1); ; seq++ { + raw, err := stdio.BusState(buscalls.ObjectsGetChunk, buscalls.ObjectsGetChunkRequest{Call: c.Call, Seq: seq}) + if err != nil { + return buscalls.ObjectData{}, fmt.Errorf("getting %s, chunk %d: %w", name, seq, err) + } + var chunk buscalls.ObjectsChunk + if err := json.Unmarshal(raw, &chunk); err != nil { + return buscalls.ObjectData{}, fmt.Errorf("the runtime's chunk %d of %s: %w", seq, name, err) + } + if chunk.Error != "" { + return buscalls.ObjectData{}, fmt.Errorf("getting %s, chunk %d: %s", name, seq, chunk.Error) + } + if chunk.Seq != seq { + return buscalls.ObjectData{}, fmt.Errorf("getting %s: chunk %d was answered as chunk %d", name, seq, chunk.Seq) + } + whole = append(whole, chunk.Data...) + if chunk.Last { + break + } + } + sum := sha256.Sum256(whole) + if c.Digest != "" && !strings.EqualFold(c.Digest, "sha256:"+hex.EncodeToString(sum[:])) { + return buscalls.ObjectData{}, fmt.Errorf("getting %s: what was read is not the value the runtime named by %s", name, c.Digest) + } + if c.Size != 0 && c.Size != uint64(len(whole)) { + return buscalls.ObjectData{}, fmt.Errorf("getting %s: %d bytes were read, the runtime said %d", name, len(whole), c.Size) + } + return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: name, Size: uint64(len(whole)), Digest: c.Digest}, Data: whole}, nil +} + +// chunkedAnswer is the session an answer to a get opens, if it is not the value itself. +func chunkedAnswer(raw json.RawMessage) (buscalls.ObjectsChunked, bool) { + var c buscalls.ObjectsChunked + if json.Unmarshal(raw, &c) != nil { + return c, false + } + switch strings.TrimSpace(string(c.Chunked)) { + case "", "null", "0", "false": + return c, false + } + return c, c.Call != "" +} diff --git a/go/busstate/session_test.go b/go/busstate/session_test.go new file mode 100644 index 00000000..a5bf2eb1 --- /dev/null +++ b/go/busstate/session_test.go @@ -0,0 +1,123 @@ +package busstate + +import ( + "bufio" + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// serveRuntime runs a bundle whose tool `go` runs fn in a goroutine and answers every call the bundle makes through handle, +// until done is closed by fn returning. +func serveRuntime(t *testing.T, fn func(), handle func(method string, params map[string]any) any) { + t.Helper() + t.Setenv(stdio.ServedAs, "distribution") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + done := make(chan struct{}) + go func() { + _ = stdio.ServeOn("", []stdio.Tool{{Name: "go", Run: func(map[string]any) (any, error) { + go func() { fn(); close(done) }() + return "started", nil + }}}, inR, outW) + }() + t.Cleanup(func() { _ = inW.Close() }) + lines := bufio.NewScanner(outR) + lines.Buffer(make([]byte, 0, 1<<20), 64<<20) + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "go"}}) + deadline := time.After(10 * time.Second) + for { + got := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + got <- m + } + }() + select { + case <-done: + return + case <-deadline: + t.Fatal("the bundle did not finish") + case m := <-got: + method, _ := m["method"].(string) + if method == "" { + continue + } + p, _ := m["params"].(map[string]any) + send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "result": handle(method, p)}) + } + } +} + +// Design 47 G §518, §345, §490 (A-21): a value larger than one chunk is put as chunks in order under one call id, each +// answered before the next, with the store named in each; a get the runtime answers as a session is read chunk by chunk and +// checked against its digest. +func TestAValueLargerThanOneChunkIsPutAndGotBySession(t *testing.T) { + value := bytes.Repeat([]byte("0123456789abcdef"), (2*buscalls.ChunkBytes+1000)/16) + sum := sha256.Sum256(value) + digest := "sha256:" + hex.EncodeToString(sum[:]) + var stored []byte + var seqs []uint32 + calls := map[string]bool{} + var gotValue []byte + var putErr, getErr, tamperErr error + var tampered bool + serveRuntime(t, func() { + o := ObjectsOf("incoming") + _, putErr = o.Put("big", value, "") + d, err := o.Get("big") + gotValue, getErr = d.Data, err + tampered = true + _, tamperErr = o.Get("big") + }, func(method string, p map[string]any) any { + switch method { + case stdio.CapabilitiesCall: + return map[string]any{"words": []string{stdio.CapPutSessions}} + case buscalls.ObjectsPutChunk: + calls[p["call"].(string)] = true + if p["objects"] != "incoming" || p["name"] != "big" { + t.Errorf("a chunk named %v in %v", p["name"], p["objects"]) + } + raw, _ := json.Marshal(p["data"]) + var b []byte + _ = json.Unmarshal(raw, &b) + stored = append(stored, b...) + seqs = append(seqs, uint32(p["seq"].(float64))) + if p["last"] == true { + return buscalls.ObjectInfo{Name: "big", Size: uint64(len(stored))} + } + return buscalls.ObjectsChunkTaken{Seq: uint32(p["seq"].(float64))} + case buscalls.ObjectsGet: + d := digest + if tampered { + d = "sha256:" + hex.EncodeToString(make([]byte, 32)) + } + return map[string]any{"chunked": 3, "size": len(stored), "digest": d, "call": "g1"} + case buscalls.ObjectsGetChunk: + seq := int(p["seq"].(float64)) + at := (seq - 1) * buscalls.ChunkBytes + return buscalls.ObjectsChunk{Seq: uint32(seq), Last: seq == 3, Data: stored[at:min(at+buscalls.ChunkBytes, len(stored))]} + } + t.Errorf("unexpected call %s", method) + return nil + }) + if putErr != nil || getErr != nil { + t.Fatalf("put %v, get %v", putErr, getErr) + } + if len(seqs) != 3 || seqs[0] != 1 || seqs[1] != 2 || seqs[2] != 3 || len(calls) != 1 || !bytes.Equal(stored, value) || !bytes.Equal(gotValue, value) { + t.Errorf("chunks %v under %d call ids, stored %d bytes, got %d", seqs, len(calls), len(stored), len(gotValue)) + } + if tamperErr == nil { + t.Error("a value that is not the one the runtime named by its digest was accepted") + } +} diff --git a/go/stateobjects/bundle.go b/go/stateobjects/bundle.go index d8ada3af..8557ad94 100644 --- a/go/stateobjects/bundle.go +++ b/go/stateobjects/bundle.go @@ -50,12 +50,9 @@ 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 refused, loudly, until the design decides how a bundle keeps one. -func (BundleObjects) MostBytes() int { return BundleObjectMost } +// 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") diff --git a/go/stateobjects/fakes_test.go b/go/stateobjects/fakes_test.go index 78dff60c..33260b24 100644 --- a/go/stateobjects/fakes_test.go +++ b/go/stateobjects/fakes_test.go @@ -2,6 +2,8 @@ package stateobjects import ( "bufio" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -25,6 +27,7 @@ type memory struct { 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{}} } @@ -216,14 +219,47 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) { 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.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.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: diff --git a/go/stateobjects/send.go b/go/stateobjects/send.go index ac0c4292..67069db1 100644 --- a/go/stateobjects/send.go +++ b/go/stateobjects/send.go @@ -8,6 +8,7 @@ import ( 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 @@ -148,11 +149,8 @@ func BundleFetch(ref Ref) ([]byte, error) { if err != nil { return nil, err } - var d buscalls.ObjectData - if err := json.Unmarshal(raw, &d); err != nil { - return nil, fmt.Errorf("the runtime's answer to %s: %w", buscalls.ObjectsSentGet, err) - } - return d.Data, nil + d, err := busstate.ReadObject(ref.Object, raw) + return d.Data, err } // ProcessFetch reads a sent object through the reader's own object store. diff --git a/go/stateobjects/send_test.go b/go/stateobjects/send_test.go index 016f3601..ea6b529f 100644 --- a/go/stateobjects/send_test.go +++ b/go/stateobjects/send_test.go @@ -41,10 +41,9 @@ func TestABodyOverTheMessageLimitIsSentAsAReference(t *testing.T) { if _, err := Receive(sent, fetch); err == nil { t.Fatal("a tampered object was accepted") } - // a bundle's store takes at most what one runtime message carries - bundle := Sender{Objects: BundleObjects{}, Store: "m_sent-objects"} - if _, err := bundle.Send("reply.huge", make([]byte, BundleObjectMost+1)); err == nil { - t.Fatal("a bundle sent a body its runtime cannot carry") + // 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") } } diff --git a/go/stateobjects/values_test.go b/go/stateobjects/values_test.go index 06d14965..a8f46de0 100644 --- a/go/stateobjects/values_test.go +++ b/go/stateobjects/values_test.go @@ -108,7 +108,7 @@ func TestValuesOfAnySizeThroughTheRuntime(t *testing.T) { m := newMemory() out := runtimeOf(t, m, func() (any, error) { v := ForState("views") - valueRows(t, v, 1, 10) // a bundle puts at most 8 MiB in an object: 100x is a process's + 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") {