novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Checker keeps a composition request in pieces, each under the state's value limit (issue 1966)

Browse Source 
mesh-host #219's check failed: the request kept for g14 was 300127 bytes compressed, over the 262144 the state takes.
The key of the check and node now holds a pointer (digest, size, pieces) and each piece of the gzip bytes is a value of
its own under pieces/<digest>/<i>, at most 128 KiB; checker_composition_requests follows the pointer and checks the digest.

Quick fix: a larger cap or a better compression: still a limit, met again as the mesh grows. Thorough fix (chosen, and
bounded only by the bucket): pieces. Not an object store: the checker is a foundation part, whose data stays in the old
sections (ADR 0619 §2), and the tool runner serves no old-section object read to its tools. No manifest change.

Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
This commit is contained in:
jschoubben
2026-11-09 16:41:56 +01:00
parent a072d60cae
commit 3f1d9cecba
3 changed files with 88 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go
+1 -1
View File
@@ -378,7 +378,7 @@ func RequestTools(r check.StateRead) []stdio.Tool {
Description: "The composition request a check sent to a node: the composer's whole input for that node (the node-engine's composition input, the case's " +
"manifests and verdicts, the seat list), kept exactly as sent so a session reproduces the check's composition (the composer run by hand over the same " +
"request gives the check's answer). Kept in the checker's state compositions, per check key (group.heads, the 12-digit heads hash) and node, " +
"about 430 KB each (kept compressed, so a value fits the state's 256 KiB), for 3 days (the state's ttl-seconds). With a check and a node, that request whole; with a check, " +
"0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for 3 days (the state's ttl-seconds). With a check and a node, that request whole; with a check, " +
"the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{
"check": map[string]any{"type": "string", "description": "the check's key, <group>.<heads hash>, as the verdict and composition name it"},
modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go
+18
View File
@@ -5,6 +5,7 @@ import (
"encoding/base64"
"encoding/json"
"errors"
"math/rand"
"os"
"regexp"
"slices"
@@ -48,7 +49,21 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
t.Fatal(err)
}
}
// Issue 1966: a request whose compressed bytes pass the state's value limit is kept in pieces and read back whole; the
// pieces are no requests in the list.
rnd := make([]byte, 600<<10)
_, _ = rand.New(rand.NewSource(1)).Read(rnd)
big := []byte(`{"input":"` + base64.StdEncoding.EncodeToString(rnd) + `"}`)
if err := keeper.Keep(context.Background(), "big.0123456789ab", "anchor", big); err != nil {
t.Fatalf("a request of %d bytes: %v", len(big), err)
}
rt := RequestTools(check.StateRead{Keys: store.Keys, Get: store.Get})
if one, err := rt[0].Run(map[string]any{"check": "big.0123456789ab", "node": "anchor"}); err != nil || !strings.Contains(func() string { b, _ := json.Marshal(one); return string(b) }(), base64.StdEncoding.EncodeToString(rnd)[:64]) {
t.Errorf("the big request read back: %v", err)
}
if all, _ := rt[0].Run(map[string]any{}); strings.Count(func() string { b, _ := json.Marshal(all); return string(b) }(), `"node"`) != 4 {
t.Errorf("the list holds a piece as a request: %v", all)
}
if len(rt) != 1 || rt[0].Name != check.RequestsReadTool || !strings.Contains(rt[0].Description, "composition request a check sent to a node") {
t.Fatalf("request tools = %+v", rt)
}
@@ -312,6 +327,9 @@ func TestTheTryToolsAskAndReadATry(t *testing.T) { // ADR 0501 §5, the agreed i
type memPut struct{ m *memStore }
func (p memPut) Put(_ context.Context, key string, value []byte) (uint64, error) {
if len(value) > 262144 { // the state's value limit, headers included (issue 1966)
return 0, errors.New("the value is over the 262144 bytes the bucket takes")
}
p.m.vals[key] = value
return 1, nil
}
modules/mesh-delivery-checker/internal/check/requests.go
+69 -18
View File
@@ -20,20 +20,28 @@ import (
// session reproduces a check's composition exactly. The checker is a foundation part (novox/hq ADR 0619 §1-3), so its
// kept data is a section of its manifest's old form, here `state`, and not an object store asked in `requires`.
//
// **What bounds it.** A request is the node-engine's whole composition input of one node, about 430 KB as JSON. A
// state's value holds at most 256 KiB (kvlimit.ModuleStateValueBytes), so it is kept gzip-compressed (JSON shrinks about
// tenfold) in a document of its own; a request whose compressed bytes still do not fit is refused by the put, which
// fails the node's composition naming the size (never dropped). The manifest's ttl-seconds is its lifetime (3 days):
// the bus drops a value past it, so nothing waits on a timer here. A check kept again overwrites the key.
// **What bounds it.** A request is the node-engine's whole composition input of one node, 0.7 to 0.8 MB as JSON and
// growing with the mesh. A state's value holds at most 256 KiB (kvlimit.ModuleStateValueBytes, headers included), and
// the compressed request already passed it (issue 1966: 300127 bytes), so it is kept gzip-compressed in pieces: the key
// of the check and node holds a small document naming the request's digest, size and number of pieces, and each piece is
// a value of its own under RequestChunkKey, of at most RequestChunkBytes. The kept data is the checker's, a foundation
// part's, so it stays in the old section `state` (ADR 0619 §2) and is read through the tool runner's state read, which
// serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (3 days) of the pointer and of
// every piece: the bus drops a value past it, so nothing waits on a timer here. A check kept again overwrites the key;
// a request whose pieces cannot all be put is refused naming the size (never dropped, never half kept: the pointer is
// put after its pieces).
const (
// RequestsStateName is the checker's state of kept requests and RequestsBucket its bucket.
RequestsStateName = "compositions"
RequestsBucket = CheckerModule + "_" + RequestsStateName
// RequestsReadTool is the tool that reads a kept request.
RequestsReadTool = "checker_composition_requests"
// RequestChunkBytes is the most compressed bytes one piece holds: base64 of it, in its document, stays well under
// the state's 256 KiB value.
RequestChunkBytes = 128 << 10
)
// KeptRequest is a request as the state holds it: whose it is, and the request's JSON, compressed.
// KeptRequest is the pointer to a request as the state holds it: whose it is, and what its pieces make.
type KeptRequest struct {
V int `json:"v"`
Check string `json:"check"`
@@ -42,8 +50,22 @@ type KeptRequest struct {
// Size and Digest are the request's, uncompressed: its bytes and their sha256.
Size int `json:"size"`
Digest string `json:"digest"`
// Gzip is the request's bytes, gzip-compressed and base64-encoded.
Gzip string `json:"gzip"`
// Pieces is how many pieces (RequestChunkKey of the digest, 0 to Pieces-1) make the request's gzip bytes.
Pieces int `json:"pieces"`
}
// KeptPiece is one piece of a request's gzip bytes, base64-encoded.
type KeptPiece struct {
V int `json:"v"`
Digest string `json:"digest"`
Index int `json:"index"`
Gzip string `json:"gzip"`
}
// RequestChunkKey is the key of a request's piece: under the request's digest, so a request kept again for the same
// bytes writes the same pieces.
func RequestChunkKey(digest string, index int) string {
return fmt.Sprintf("pieces/%s/%d", digest, index)
}
// RequestKey is the key a check's request to a node is kept under: the check's key (group.heads) and the node.
@@ -75,14 +97,28 @@ func (k RequestKeeper) Keep(ctx context.Context, key, node string, body []byte)
now = k.Now
}
sum := sha256.Sum256(body)
doc, err := json.Marshal(KeptRequest{V: 1, Check: key, Node: node, KeptAt: now().UTC(), Size: len(body),
Digest: hex.EncodeToString(sum[:]), Gzip: base64.StdEncoding.EncodeToString(packed.Bytes())})
digest := hex.EncodeToString(sum[:])
pieces := (packed.Len() + RequestChunkBytes - 1) / RequestChunkBytes
fail := func(err error) error {
return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed in %d pieces) could not be kept in the state %s: %w",
key, node, len(body), packed.Len(), pieces, RequestsStateName, err)
}
for i := 0; i < pieces; i++ {
end := min((i+1)*RequestChunkBytes, packed.Len())
doc, err := json.Marshal(KeptPiece{V: 2, Digest: digest, Index: i, Gzip: base64.StdEncoding.EncodeToString(packed.Bytes()[i*RequestChunkBytes : end])})
if err != nil {
return err
}
if _, err := k.Store.Put(ctx, RequestChunkKey(digest, i), doc); err != nil {
return fail(err)
}
}
doc, err := json.Marshal(KeptRequest{V: 2, Check: key, Node: node, KeptAt: now().UTC(), Size: len(body), Digest: digest, Pieces: pieces})
if err != nil {
return err
}
if _, err := k.Store.Put(ctx, RequestKey(key, node), doc); err != nil {
return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed) could not be kept in the state %s: %w",
key, node, len(body), packed.Len(), RequestsStateName, err)
return fail(err)
}
return nil
}
@@ -104,7 +140,7 @@ func ReadRequests(s StateRead, key, node string) (any, error) {
if len(raw) == 0 {
return nil, fmt.Errorf("no composition request of check %s to %s is kept (none was sent, or it is past its lifetime)", key, node)
}
kept, body, err := unpackRequest(raw)
kept, body, err := unpackRequest(s, raw)
if err != nil {
return nil, fmt.Errorf("the kept request %s: %w", RequestKey(key, node), err)
}
@@ -129,20 +165,35 @@ func ReadRequests(s StateRead, key, node string) (any, error) {
listed = append(listed, map[string]any{"key": k, "unreadable": err.Error()})
continue
}
if kept.Check == "" {
continue // a piece of a request, not a request
}
listed = append(listed, map[string]any{"check": kept.Check, "node": kept.Node, "size": kept.Size, "digest": kept.Digest, "kept_at": kept.KeptAt})
}
return map[string]any{"kept": listed}, nil
}
// unpackRequest reads a kept document and its request's bytes, which must match the digest they were kept with.
func unpackRequest(raw []byte) (KeptRequest, []byte, error) {
// unpackRequest reads a kept pointer's pieces and the request's bytes, which must match the digest they were kept with.
func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) {
var kept KeptRequest
if err := json.Unmarshal(raw, &kept); err != nil {
return kept, nil, fmt.Errorf("is no kept request: %w", err)
}
packed, err := base64.StdEncoding.DecodeString(kept.Gzip)
if err != nil {
return kept, nil, fmt.Errorf("holds no base64: %w", err)
var packed []byte
for i := 0; i < kept.Pieces; i++ {
raw, err := s.Get(RequestChunkKey(kept.Digest, i))
if err != nil {
return kept, nil, err
}
var piece KeptPiece
if err := json.Unmarshal(raw, &piece); err != nil || len(raw) == 0 {
return kept, nil, fmt.Errorf("lacks its piece %d of %d (past its lifetime, or never kept)", i, kept.Pieces)
}
part, err := base64.StdEncoding.DecodeString(piece.Gzip)
if err != nil {
return kept, nil, fmt.Errorf("holds no base64 in its piece %d: %w", i, err)
}
packed = append(packed, part...)
}
zr, err := gzip.NewReader(bytes.NewReader(packed))
if err != nil {