Merge commit '7633a2fe' into feat/47-G-fix11-b5g4

modules/mesh-delivery-checker/cmd/mesh-checking-program/main_test.go

@@ -153,8 +153,8 @@ func TestTheCheckingProgramAnswersTheVerdictOfItsFacts(t *testing.T) {
 	}
 	// Review finding 3 of milestone 47/G: the heads of a real group carry every seat definition the reader holds, unchanged
 	// (this repository's own contracts, laid as its head): the answer leaves them all to the checker's list and stays
-	// within the bound a run's answer is written under (buildstep.RunInputBound), where the whole list would not.
-	t.Run("heads carrying every definition the reader holds unchanged answer none of them, within the bound", func(t *testing.T) {
+	// far below the whole list.
+	t.Run("heads carrying every definition the reader holds unchanged answer none of them", func(t *testing.T) {
 		real := under{prefix: "novox/mesh-delivery", fs: os.DirFS("../../../..")}
 		group := rules.Facts{V: rules.FactsVersion, Members: []rules.Member{{Repository: "novox/mesh-delivery", Head: "h"}}}
 		// An empty list of digests answers every definition whole (a checker that hands its list's digests, holding none).
@@ -166,9 +166,9 @@ func TestTheCheckingProgramAnswersTheVerdictOfItsFacts(t *testing.T) {
 		group.SeatDigests = wholeDigests(t, real, "novox/mesh-delivery")
 		same, err := rules.SeatsOf(real, group)
 		answer, _ := json.Marshal(same)
-		if err != nil || len(same.Seats) != 0 || same.Same != len(whole.Seats) || len(answer) > buildstep.RunInputBound {
-			t.Errorf("unchanged heads answer %d definitions (%d the same), %d bytes; want none, all %d the same, within %d: %v",
-				len(same.Seats), same.Same, len(answer), len(whole.Seats), buildstep.RunInputBound, err)
+		if err != nil || len(same.Seats) != 0 || same.Same != len(whole.Seats) {
+			t.Errorf("unchanged heads answer %d definitions (%d the same), %d bytes; want none, all %d the same: %v",
+				len(same.Seats), same.Same, len(answer), len(whole.Seats), err)
 		}
 		// Review round 2 of G: a checker from before G (B0's, which makes G's own check) hands no digests, and the answer
 		// then keeps each seat's own description only, so a group's whole list stays within the bound; each still parses.
@@ -206,8 +206,8 @@ func TestTheCheckingProgramAnswersTheVerdictOfItsFacts(t *testing.T) {
 				t.Errorf("B0-shaped facts answer %s with %d inner descriptions, its own %q: %v; want only its own", raw[:min(len(raw), 80)], inner, d.Description, perr)
 			}
 		}
-		if answer, _ := json.Marshal(b0); len(answer) > buildstep.RunInputBound || len(answer) >= len(mustJSON(t, whole))*3/4 {
-			t.Errorf("B0-shaped facts answer %d bytes (whole %d), want within %d and far below the whole", len(answer), len(mustJSON(t, whole)), buildstep.RunInputBound)
+		if answer, _ := json.Marshal(b0); len(answer) >= len(mustJSON(t, whole))*3/4 {
+			t.Errorf("B0-shaped facts answer %d bytes (whole %d), want far below the whole", len(answer), len(mustJSON(t, whole)))
 		}
 		// One definition the heads change is answered alone.
 		for seat := range group.SeatDigests {

modules/mesh-delivery-checker/rules/seats.go

@@ -21,8 +21,8 @@ import (
 // milestones 47/B0 and 47/G): the checking program built from the heads reads the definitions the heads' contracts
 // modules carry, by the domain SDK's one reading (sdk/seatdefs), the indexer's too, and answers them; the running checker
 // lays them by seat over the list its reader holds from the seat mesh-catalog (verb seat-definitions) and reads no file
-// of the heads. The reader's list is not in the program's input: the mesh's definitions are larger than a run input's
-// bound (buildstep.RunInputBound).
+// of the heads. The reader's list is not in the program's input: it is large, and the answer carries only what the heads
+// change.
 
 // SeatsVersion is the version of SeatsAnswer this program writes.
 const SeatsVersion = 1

modules/mesh-delivery-contracts/go/seats/meshcatalog/contract.go

@@ -1379,16 +1379,18 @@ type OutdatedAnswer struct {
 
 // InputInput is what input takes.
 type InputInput struct {
+	// 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements
+	Call string `json:"call"`
 	// the input's sha256
 	Digest string `json:"digest"`
 }
 
 // InputAnswer is what input answers.
 type InputAnswer struct {
+	// how many chunks are sent; 0 when there is no input of that digest
+	Chunks int64 `json:"chunks"`
 	// the input's sha256
 	Digest string `json:"digest"`
-	// the input's exact bytes, base64
-	Input string `json:"input"`
 }
 
 // EntryInput is what entry takes.

modules/mesh-delivery-contracts/go/seats/meshcatalog/contract_test.go

@@ -171,9 +171,9 @@ func TestFixtures(t *testing.T) {
 			decode: func(b []byte) (any, error) { var v OutdatedInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
 		{name: "outdated answers an entry behind", subject: "", given: "{\"behind\":[{\"act\":\"rebuild\",\"dependency\":{\"artifact\":\"mesh-sdk:go\",\"branch\":\"main\",\"source\":3589},\"kind\":1,\"module\":\"nats\",\"newest\":\"0.1.15 sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"recorded\":\"0.1.14 sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"version\":\"1.2.3\"}]}", wire: "{\"behind\":[{\"act\":\"rebuild\",\"dependency\":{\"artifact\":\"mesh-sdk:go\",\"branch\":\"main\",\"source\":3589},\"kind\":1,\"module\":\"nats\",\"newest\":\"0.1.15 sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"recorded\":\"0.1.14 sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"version\":\"1.2.3\"}]}",
 			decode: func(b []byte) (any, error) { var v OutdatedAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
-		{name: "input asks a step's input by digest", subject: "", given: "{\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}", wire: "{\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}",
+		{name: "input asks a step's input by digest", subject: "", given: "{\"call\":\"0123456789abcdef0123456789abcdef\",\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}", wire: "{\"call\":\"0123456789abcdef0123456789abcdef\",\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}",
 			decode: func(b []byte) (any, error) { var v InputInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
-		{name: "input answers the exact bytes", subject: "", given: "{\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"input\":\"eyJ2IjoxfQ==\"}", wire: "{\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\",\"input\":\"eyJ2IjoxfQ==\"}",
+		{name: "input answers how many chunks carry the exact bytes", subject: "", given: "{\"chunks\":3,\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}", wire: "{\"chunks\":3,\"digest\":\"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}",
 			decode: func(b []byte) (any, error) { var v InputAnswer; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},
 		{name: "entry asks by fingerprint", subject: "", given: "{\"fingerprint\":\"src2:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}", wire: "{\"fingerprint\":\"src2:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\"}",
 			decode: func(b []byte) (any, error) { var v EntryInput; err := json.Unmarshal(b, &v); return v, err }, subjectIs: nil},

modules/mesh-delivery-contracts/seats/mesh-catalog/fixtures/verbs.json

@@ -424,15 +424,16 @@
     "name": "input asks a step's input by digest",
     "of": "verb:input.input",
     "given": {
+      "call": "0123456789abcdef0123456789abcdef",
       "digest": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
     }
   },
   {
-    "name": "input answers the exact bytes",
+    "name": "input answers how many chunks carry the exact bytes",
     "of": "verb:input.answer",
     "given": {
-      "digest": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
-      "input": "eyJ2IjoxfQ=="
+      "chunks": 3,
+      "digest": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
     }
   },
   {

modules/mesh-delivery-contracts/seats/mesh-catalog/seat.json

@@ -171,7 +171,7 @@
     },
     {
       "name": "seat-definitions",
-      "description": "The seat list (ADR 0636 \u00a74): with no seat, every trunk seat with its digest and the revision of the list; with a seat, that trunk definition resolved with the parts it includes. Reads only.",
+      "description": "The seat list (ADR 0636 §4): with no seat, every trunk seat with its digest and the revision of the list; with a seat, that trunk definition resolved with the parts it includes. Reads only.",
       "input": {
         "properties": {
           "seat": {
@@ -250,7 +250,7 @@
     },
     {
       "name": "revision",
-      "description": "The catalog half of the revision view (ADR 0656 \u00a76): one catalog graph revision with its entries whole, the trunk heads, the seat definitions by digest, and each manifest's verdict and fingerprint; the trunk's newest when no revision and no candidate is named, or a candidate's for its group and heads. Answered by the holder from what it holds, in its own runtime (ADR 0663), in pages of entries. Reads only. A watched branch with no revision yet answers {none-yet: true} alone (design 47 G section 530).",
+      "description": "The catalog half of the revision view (ADR 0656 §6): one catalog graph revision with its entries whole, the trunk heads, the seat definitions by digest, and each manifest's verdict and fingerprint; the trunk's newest when no revision and no candidate is named, or a candidate's for its group and heads. Answered by the holder from what it holds, in its own runtime (ADR 0663), in pages of entries. Reads only. A watched branch with no revision yet answers {none-yet: true} alone (design 47 G section 530).",
       "input": {
         "type": "object",
         "properties": {
@@ -554,17 +554,23 @@
         "seat:node-build-node",
         "seat:node-build-image"
       ],
-      "description": "A step's input by its digest, as the indexer keeps it for a validation step it asked a build seat to run (design 47 G section 354).",
+      "description": "A step's input by its digest, as the indexer keeps it for a validation step it asked a build seat to run (design 47 G section 354): its fields, then the bytes in chunks of at most 512 KiB on the chunk subject of the call, each acknowledged by the caller on the ack subject within 30 seconds; the input is of any size.",
       "input": {
         "type": "object",
         "properties": {
           "digest": {
             "type": "string",
             "description": "the input's sha256"
+          },
+          "call": {
+            "type": "string",
+            "description": "128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements",
+            "minLength": 1
           }
         },
         "required": [
-          "digest"
+          "digest",
+          "call"
         ]
       },
       "answer": {
@@ -574,16 +580,18 @@
             "type": "string",
             "description": "the input's sha256"
           },
-          "input": {
-            "type": "string",
-            "description": "the input's exact bytes, base64"
+          "chunks": {
+            "type": "integer",
+            "minimum": 0,
+            "description": "how many chunks are sent; 0 when there is no input of that digest"
           }
         },
         "required": [
           "digest",
-          "input"
+          "chunks"
         ]
-      }
+      },
+      "chunks": true
     },
     {
       "name": "entry",
@@ -614,7 +622,7 @@
       }
     }
   ],
-  "description": "The catalog of the mesh: every module the trunk holds with its manifest, fingerprint and verdict, the packages published, and the seat list (ADR 0636 \u00a71, \u00a74).",
+  "description": "The catalog of the mesh: every module the trunk holds with its manifest, fingerprint and verdict, the packages published, and the seat list (ADR 0636 §1, §4).",
   "family": {
     "name": "mesh-catalog.events",
     "max-age-days": 7
@@ -1793,7 +1801,7 @@
       "name": "inputs",
       "template": "state",
       "history": 1,
-      "description": "a step's input, named by its write time and its sha256 (design 47 G section 354)",
+      "description": "a step's input, named by its write time and its sha256; a value over one chunk is held in the state's objects store (design 47 G sections 354, 712)",
       "key-forms": [
         {
           "key": "<time>.<digest>",
@@ -1918,7 +1926,7 @@
           "additionalProperties": {
             "$ref": "#/defs/BuildEdge"
           },
-          "description": "its build edges, by language (ADR 0603 \u00a72)"
+          "description": "its build edges, by language (ADR 0603 §2)"
         },
         "provides": {
           "type": "array",
@@ -1958,14 +1966,14 @@
           "items": {
             "type": "string"
           },
-          "description": "its verb edges left out because they lie on a cycle (ADR 0614 \u00a73)"
+          "description": "its verb edges left out because they lie on a cycle (ADR 0614 §3)"
         },
         "needs": {
           "type": "array",
           "items": {
             "type": "string"
           },
-          "description": "the manifest reads a node-engine must state to compose it, its templates' forms included, as the forge bridge read them at the merge (ADR 0614 \u00a74, ADR 0632 \u00a72)"
+          "description": "the manifest reads a node-engine must state to compose it, its templates' forms included, as the forge bridge read them at the merge (ADR 0614 §4, ADR 0632 §2)"
         },
         "main": {
           "type": "string",
@@ -2256,7 +2264,7 @@
     },
     "BuildEdge": {
       "type": "object",
-      "description": "the build edge of one language: the build module that builds the module, and the toolchain and recipe version its fingerprint names (ADR 0603 \u00a72)",
+      "description": "the build edge of one language: the build module that builds the module, and the toolchain and recipe version its fingerprint names (ADR 0603 §2)",
       "properties": {
         "module": {
           "type": "string",

modules/mesh-delivery-contracts/ts/seats/meshcatalog/index.d.ts

@@ -829,16 +829,18 @@ export interface OutdatedAnswer {
 
 /** is what input takes. */
 export interface InputInput {
+  /** 128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements */
+  "call": string;
   /** the input's sha256 */
   "digest": string;
 }
 
 /** is what input answers. */
 export interface InputAnswer {
+  /** how many chunks are sent; 0 when there is no input of that digest */
+  "chunks": number;
   /** the input's sha256 */
   "digest": string;
-  /** the input's exact bytes, base64 */
-  "input": string;
 }
 
 /** is what entry takes. */

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
 	"os"
 	"strings"
 	"sync"
@@ -291,7 +292,7 @@ func main() {
 	// the parts whose entry declares it (novox/hq ADR 0663).
 	tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}),
 		index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}, Revisions: revisions}),
-		index.InputTool(keptRead{stdio.State(index.InputsState)}), index.InputsReadTool(keptRead{stdio.State(index.InputsState)}))
+		index.InputsReadTool(keptRead{stdio.State(index.InputsState)}, stateobjects.ForState(index.InputsState).Objects))
 	tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error {
 		_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"})
 		return err

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go

@@ -0,0 +1,95 @@
+//go:build integration
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
+	"mesh-delivery-indexer/internal/index"
+)
+
+// TestIndexerAnswersARunsInputOverOneMiBInChunks (design 47 G §712, A-237, §681): an input of 3 MiB, kept in the state inputs with
+// its object beside it, is answered by the seat mesh-catalog's verb input as a count of chunks and then sent on the call's chunk
+// subjects, each acknowledged by the caller; the chunks carry the exact bytes.
+func TestIndexerAnswersARunsInputOverOneMiBInChunks(t *testing.T) {
+	body := make([]byte, 3<<20)
+	if _, err := rand.Read(body); err != nil {
+		t.Fatal(err)
+	}
+	sum := sha256.Sum256(body)
+	digest := hex.EncodeToString(sum[:])
+	js, _, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.InputsBucket, History: 1})
+		if err != nil {
+			t.Fatal(err)
+		}
+		objs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))})
+		if err != nil {
+			t.Fatal(err)
+		}
+		key := meshcatalog.InputsByTimeDigestKey(time.Now().UTC().Format("20060102T150405Z"), digest)
+		if err := stateobjects.OverBus(kv, objs, catalogue.StateObjectsName(index.InputsState)).Put(key, base64.StdEncoding.EncodeToString(body)); err != nil {
+			t.Fatal(err)
+		}
+	})
+	nc := js.Conn()
+	call := "0123456789abcdef0123"
+	send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: call}
+	chunks := make(chan *nats.Msg, 16)
+	sub, err := nc.ChanSubscribe(send.Chunks(), chunks)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.Unsubscribe()
+	req, _ := json.Marshal(meshcatalog.InputInput{Digest: digest, Call: call})
+	var reply struct {
+		Result meshcatalog.InputAnswer `json:"result"`
+		Error  string                  `json:"error"`
+	}
+	for deadline := time.Now().Add(30 * time.Second); ; time.Sleep(100 * time.Millisecond) {
+		m, err := nc.Request("mesh.seat.mesh-catalog.tool.input", req, 5*time.Second)
+		if err == nil {
+			if err := json.Unmarshal(m.Data, &reply); err != nil {
+				t.Fatal(err)
+			}
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("the verb input is not answered within 30 s: %v", err)
+		}
+	}
+	if reply.Error != "" || reply.Result.Digest != digest || reply.Result.Chunks != 6 {
+		t.Fatalf("the answer %+v (%q), want %s in 6 chunks", reply.Result, reply.Error, digest)
+	}
+	var got bytes.Buffer
+	next := func() (string, []byte, error) {
+		select {
+		case m := <-chunks:
+			return m.Subject, m.Data, nil
+		case <-time.After(20 * time.Second):
+			return "", nil, buscalls.ErrSeatChunkMissed
+		}
+	}
+	ack := func(subject string) error { _, err := nc.Request(subject, []byte("{}"), 5*time.Second); return err }
+	if _, err := send.Receive(reply.Result.Chunks, next, ack, &got); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(got.Bytes(), body) {
+		t.Errorf("the chunks carry %d bytes that are not the input's %d", got.Len(), len(body))
+	}
+}

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
+	"mesh-delivery-indexer/internal/index"
+)
+
+// inputObjects is the object store beside the state inputs (`inputs-objects`, a state's own, design 47 G §712, A-237), opened when
+// first used: until the grants part has made it every call is process.ErrNotYet.
+type inputObjects struct {
+	js     jetstream.JetStream
+	module string
+	mu     sync.Mutex
+	bus    stateobjects.Objects
+}
+
+func (o *inputObjects) open() (stateobjects.Objects, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.bus != nil {
+		return o.bus, nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	store, err := o.js.ObjectStore(ctx, catalogue.BusName(o.module, catalogue.StateObjectsName(index.InputsState)))
+	if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil, fmt.Errorf("the object store of the state inputs is not on the bus yet: %w", process.ErrNotYet)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the object store of the state inputs is not open: %w", err)
+	}
+	o.bus = stateobjects.BusObjects{Store: store}
+	return o.bus, nil
+}
+
+func (o *inputObjects) MostBytes() int { return 0 }
+
+func (o *inputObjects) Put(name string, data []byte) error {
+	b, err := o.open()
+	if err != nil {
+		return err
+	}
+	return b.Put(name, data)
+}
+
+func (o *inputObjects) Get(name string) ([]byte, error) {
+	b, err := o.open()
+	if err != nil {
+		return nil, err
+	}
+	return b.Get(name)
+}
+
+func (o *inputObjects) Delete(name string) error {
+	b, err := o.open()
+	if err != nil {
+		return err
+	}
+	return b.Delete(name)
+}
+
+func (o *inputObjects) List() ([]stateobjects.Object, error) {
+	b, err := o.open()
+	if err != nil {
+		return nil, err
+	}
+	return b.List()
+}
+
+// coreChunks is the core connection as a send of chunks reaches it (index.ChunkLine): a chunk is published on no stream, an
+// acknowledgement is a request the holder answers.
+type coreChunks struct{ nc *nats.Conn }
+
+func (c coreChunks) Publish(subject string, data []byte) error { return c.nc.Publish(subject, data) }
+
+func (c coreChunks) Hear(subject string, fn func(string)) (func(), error) {
+	sub, err := c.nc.Subscribe(subject, func(m *nats.Msg) {
+		fn(m.Subject)
+		if m.Reply != "" {
+			_ = m.Respond([]byte("{}"))
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	return func() { _ = sub.Unsubscribe() }, nil
+}

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -23,9 +23,11 @@ import (
 	"time"
 
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -443,6 +445,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	pinsBucket := process.AwaitBucket(ctx, js, conn, index.PinsBucket, "no record's pin on a catalog graph revision is kept, so no revision is removed", logger.Printf)
 	pinsStore := awaitedStore{b: pinsBucket}
 	consumers := process.AwaitBucket(ctx, js, conn, index.Module+"_"+process.ConsumersState, "indexer_consumer reads no consumer", logger.Printf)
+	// The state inputs and the object store beside it; the seat mesh-catalog's verb input is answered here in chunks of the core
+	// connection, of any size (design 47 G §712, A-237).
+	inputsStore := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)}
+	inObjects := &inputObjects{js: js, module: cred.Module}
+	stopInput, err := calls.AnswerSeat(meshcatalog.Seat, map[string]process.SeatHandler{
+		meshcatalog.VerbInput: index.InputVerb(inputsStore, inObjects, coreChunks{conn}, buscalls.SeatAckWithin, logger.Printf)})
+	if err != nil {
+		return fmt.Errorf("answering the verb input: %w", err)
+	}
+	defer stopInput()
 	ix := &index.Indexer{NamedSets: named, Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known,
 		Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred),
 		Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf,
@@ -458,7 +470,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Versions:   awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)},
 		BuildEdges: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.BuildEdgesBucket, "no build of another watched branch keeps its edges", logger.Printf)},
 		Pins:       pinsStore, PinsPending: pendingOf(js, cred),
-		Inputs:             awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)},
+		Inputs:             inputsStore,
+		InputObjects:       inObjects,
 		SayVersionReleased: sayVersionReleased(conn, cred), SayVersionsFilled: sayVersionsFilled(conn, cred),
 		// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
 		// until then every merge is taken in at once, as before.

modules/mesh-delivery-indexer/internal/index/candidates.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
 	"reflect"
 	"slices"
 	"strings"
@@ -120,6 +121,9 @@ type Indexer struct {
 	SayVersionsFilled func(ctx context.Context, e meshcatalog.VersionsFilledEvent, id string) error
 	// Inputs is the state inputs (run_inputs.go, design 47 G §354): the inputs of the indexer's run asks; nil keeps none.
 	Inputs Store
+	// InputObjects is the object store beside the state inputs (`inputs-objects`): an input too large for a key is kept there
+	// (mesh-sdk stateobjects, design 47 G §712, A-237); nil keeps a large input nowhere and refuses it.
+	InputObjects stateobjects.Objects
 	// Registry is the registrations and their trusted changes (registrations.go, design 47 G §2 to §4, §667); SeedModules reads
 	// what the first registrations are composed from (§8). Nil: no registration act is answered.
 	Registry    *Registry

modules/mesh-delivery-indexer/internal/index/pending.go

@@ -910,7 +910,7 @@ func (ix *Indexer) dry() *dryRun {
 		Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
 		OwnBuildState: nil, Heard: nil, Placement: nil, RunningWhole: nil, underSeats: true,
 		// A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep.
-		Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil, Pins: nil, PinsPending: nil,
+		Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil, InputObjects: nil, Pins: nil, PinsPending: nil,
 		// A dry run writes no registration and asks no one.
 		Registry: nil, SeedModules: nil}
 	x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }

modules/mesh-delivery-indexer/internal/index/pending_test.go

@@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
 			"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
 			"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn",
-			"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges", "Pins", "PinsPending"} {
+			"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "InputObjects", "BuildEdges", "Pins", "PinsPending"} {
 			named[f] = true
 		}
 		ty := reflect.TypeOf(Indexer{})

modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go

@@ -0,0 +1,132 @@
+package index
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
+)
+
+type objectsMem struct {
+	mu   sync.Mutex
+	objs map[string][]byte
+}
+
+func (*objectsMem) MostBytes() int { return 0 }
+func (o *objectsMem) Put(name string, data []byte) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.objs[name] = data
+	return nil
+}
+func (o *objectsMem) Get(name string) ([]byte, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	d, ok := o.objs[name]
+	if !ok {
+		return nil, errors.New("no such object")
+	}
+	return d, nil
+}
+func (o *objectsMem) Delete(name string) error             { return nil }
+func (o *objectsMem) List() ([]stateobjects.Object, error) { return nil, nil }
+
+// lineMem is a core bus in memory: a chunk published is kept, and its acknowledgement heard at once.
+type lineMem struct {
+	mu     sync.Mutex
+	hear   func(string)
+	chunks map[string][]byte
+}
+
+func (l *lineMem) Publish(subject string, data []byte) error {
+	l.mu.Lock()
+	l.chunks[subject] = data
+	h := l.hear
+	l.mu.Unlock()
+	var c buscalls.SeatChunk
+	if err := json.Unmarshal(data, &c); err != nil {
+		return err
+	}
+	go h(buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: "0123456789abcdef"}.Ack(c.N))
+	return nil
+}
+
+func (l *lineMem) Hear(_ string, fn func(string)) (func(), error) {
+	l.mu.Lock()
+	l.hear = fn
+	l.mu.Unlock()
+	return func() {}, nil
+}
+
+// Design 47 G §712, A-237: an input is of any size: over a key's limit it is kept as an object beside the state, and the verb
+// input answers it in chunks of at most 512 KiB, each acknowledged before the next.
+func TestAnInputOverOneMiBIsKeptAsAnObjectAndAnsweredInChunks(t *testing.T) {
+	ctx := context.Background()
+	objs := &objectsMem{objs: map[string][]byte{}}
+	ix := &Indexer{Inputs: newMem(), InputObjects: objs, Now: time.Now}
+	body := make([]byte, 3<<20+17)
+	if _, err := rand.Read(body); err != nil {
+		t.Fatal(err)
+	}
+	digest, err := ix.KeepInput(ctx, "ask-big", body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(objs.objs) != 1 {
+		t.Fatalf("a large input is one object beside the state, got %d", len(objs.objs))
+	}
+	if got, err := InputBody(ctx, ix.Inputs, objs, digest); err != nil || !bytes.Equal(got, body) {
+		t.Fatalf("the input read back: %d bytes, %v", len(got), err)
+	}
+	line := &lineMem{chunks: map[string][]byte{}}
+	verb := InputVerb(ix.Inputs, objs, line, time.Second, t.Logf)
+	raw, _ := json.Marshal(meshcatalog.InputInput{Digest: digest, Call: "0123456789abcdef"})
+	out, err := verb("", raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ans := out.(meshcatalog.InputAnswer)
+	want := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes)
+	if ans.Digest != digest || ans.Chunks != want || want < 6 {
+		t.Fatalf("the answer %+v, want %d chunks", ans, want)
+	}
+	send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: "0123456789abcdef"}
+	var got []byte
+	deadline := time.Now().Add(5 * time.Second)
+	for n := int64(1); n <= want; n++ {
+		for {
+			line.mu.Lock()
+			data, ok := line.chunks[send.Chunk(n)]
+			line.mu.Unlock()
+			if ok {
+				var c buscalls.SeatChunk
+				if err := json.Unmarshal(data, &c); err != nil || c.N != n || c.Of != want {
+					t.Fatalf("chunk %d: %+v, %v", n, c, err)
+				}
+				got = append(got, c.Data...)
+				break
+			}
+			if time.Now().After(deadline) {
+				t.Fatalf("chunk %d of %d never sent", n, want)
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}
+	if !bytes.Equal(got, body) {
+		t.Error("the chunks do not carry the input's bytes")
+	}
+	if _, err := verb("", json.RawMessage(`{"digest":"sha256:none","call":"0123456789abcdef"}`)); err == nil {
+		t.Error("an input never kept is answered")
+	}
+	if _, err := verb("", json.RawMessage(`{"digest":"`+digest+`","call":"x"}`)); err == nil {
+		t.Error("a call of no token is answered")
+	}
+}

modules/mesh-delivery-indexer/internal/index/run_inputs.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"regexp"
 	"slices"
 	"strings"
 	"time"
@@ -15,7 +16,10 @@ import (
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/keys"
+	"git.novox.be/novox/mesh-sdk/go/stateobjects"
 )
 
 // The indexer's inputs (design 47 G §354): the input of the ask `run` the indexer's validation puts, kept in the old state
@@ -42,6 +46,13 @@ func InputDigest(body []byte) string {
 	return "sha256:" + hex.EncodeToString(sum[:])
 }
 
+// runDigest is the digest a run ask names its input by (buildstep.RunInput.Digest): the 64 hex digits of its sha256, which are also
+// the digest of the key's form, a name a state's key may hold (a ':' may not be).
+func runDigest(body []byte) string {
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:])
+}
+
 // KeepInput keeps the exact bytes of the input of the run ask ask under their write time and sha256, after sweeping the keys more
 // than a day old among the 20 oldest, and names it by the ask's id (A-78 d); it answers the digest. The same bytes written again in
 // the same second are one key.
@@ -55,13 +66,9 @@ func (ix *Indexer) KeepInput(ctx context.Context, ask string, body []byte) (stri
 	if err := ix.sweepInputs(ctx); err != nil {
 		ix.logf("inputs: the keys older than a day are not all deleted: %v", err)
 	}
-	digest := InputDigest(body)
+	digest := runDigest(body)
 	key := meshcatalog.InputsByTimeDigestKey(ix.now().UTC().Format(inputsTime), digest)
-	data, err := json.Marshal(base64.StdEncoding.EncodeToString(body))
-	if err != nil {
-		return "", err
-	}
-	if err := ix.Inputs.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) {
+	if err := inputValues(ix.Inputs, ix.InputObjects).Put(key, base64.StdEncoding.EncodeToString(body)); err != nil && !errors.Is(err, stateobjects.ErrChanged) {
 		return "", err
 	}
 	if err := ix.Inputs.Create(ctx, InputsAskPrefix+keys.EscapeName(ask), []byte(key)); err != nil && !errors.Is(err, ErrConflict) {
@@ -149,40 +156,132 @@ func InputKey(ctx context.Context, st Store, digest string) (string, error) {
 	return "", nil
 }
 
-// Input answers the verb input: the exact bytes (base64) of the input of the digest, its digest checked.
-func Input(ctx context.Context, st Store, in meshcatalog.InputInput) (meshcatalog.InputAnswer, error) {
-	var out meshcatalog.InputAnswer
-	if in.Digest == "" {
-		return out, errors.New("name the input's sha256 (digest)")
-	}
-	key, err := InputKey(ctx, st, in.Digest)
-	if err != nil {
-		return out, fmt.Errorf("the state %s cannot be read: %w", InputsState, err)
-	}
-	raw, _, found, err := st.Get(ctx, key)
-	if key == "" || err != nil || !found {
-		return out, fmt.Errorf("no input of digest %s is kept", in.Digest)
-	}
-	text, err := meshcatalog.DecodeInputsByTimeDigestValue(raw)
-	if err != nil {
-		return out, fmt.Errorf("the input %s cannot be read: %w", key, err)
-	}
-	body, err := base64.StdEncoding.DecodeString(text)
-	if err != nil || InputDigest(body) != in.Digest {
-		return out, fmt.Errorf("the input kept under %s does not match its digest", key)
-	}
-	return meshcatalog.InputAnswer{Digest: in.Digest, Input: text}, nil
+// inputValues is the state inputs with its object store beside it (mesh-sdk stateobjects): a value that fits a key is in it, a
+// larger one is an object with its reference in the key (ADR 0666 §16).
+func inputValues(st Store, objs stateobjects.Objects) stateobjects.Values {
+	return stateobjects.Values{KV: storeKV{st}, Objects: objs, Store: catalogue.StateObjectsName(InputsState)}
 }
 
-// InputTool is the verb input as the tool mesh-catalog.input of the indexer's own runtime.
-func InputTool(st Store) stdio.Tool {
-	return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbInput, Run: func(args map[string]any) (any, error) {
+// storeKV is a Store as the state's key-value side.
+type storeKV struct{ s Store }
+
+func (k storeKV) Get(key string) (*stateobjects.Entry, error) {
+	v, rev, found, err := k.s.Get(context.Background(), key)
+	if err != nil || !found {
+		return nil, err
+	}
+	return &stateobjects.Entry{Value: v, Revision: rev}, nil
+}
+
+func changed(err error) error {
+	if errors.Is(err, ErrConflict) {
+		return stateobjects.ErrChanged
+	}
+	return err
+}
+
+func (k storeKV) Create(key string, v []byte) error {
+	return changed(k.s.Create(context.Background(), key, v))
+}
+
+func (k storeKV) Update(key string, v []byte, rev uint64) error {
+	return changed(k.s.Update(context.Background(), key, v, rev))
+}
+
+func (k storeKV) Delete(key string) error {
+	_, rev, found, err := k.s.Get(context.Background(), key)
+	if err != nil || !found {
+		return err
+	}
+	return k.s.Delete(context.Background(), key, rev)
+}
+
+func (k storeKV) Keys() ([]string, error) { return k.s.Keys(context.Background()) }
+
+// InputBody is the exact bytes of the input of the digest, checked against it, whatever its size.
+func InputBody(ctx context.Context, st Store, objs stateobjects.Objects, digest string) ([]byte, error) {
+	if digest == "" {
+		return nil, errors.New("name the input's sha256 (digest)")
+	}
+	key, err := InputKey(ctx, st, digest)
+	if err != nil {
+		return nil, fmt.Errorf("the state %s cannot be read: %w", InputsState, err)
+	}
+	if key == "" {
+		return nil, fmt.Errorf("no input of digest %s is kept", digest)
+	}
+	var text string
+	found, err := inputValues(st, objs).Get(key, &text)
+	if err != nil {
+		return nil, fmt.Errorf("the input %s cannot be read: %w", key, err)
+	}
+	if !found {
+		return nil, fmt.Errorf("no input of digest %s is kept", digest)
+	}
+	body, err := base64.StdEncoding.DecodeString(text)
+	if err != nil || runDigest(body) != digest {
+		return nil, fmt.Errorf("the input kept under %s does not match its digest", key)
+	}
+	return body, nil
+}
+
+// ChunkLine is the core bus as a send of chunks reaches it: a chunk is published on its subject, kept on no stream, and each is
+// acknowledged by the caller as a request on its ack subject, which the holder answers (design 47 G §681).
+type ChunkLine interface {
+	Publish(subject string, data []byte) error
+	// Hear calls fn with the subject of each request made on subject (a pattern) and answers it, until stop is called.
+	Hear(subject string, fn func(subject string)) (stop func(), err error)
+}
+
+var callToken = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`)
+
+// InputVerb answers the verb input (marked chunks): the digest and how many chunks carry the bytes, then sends them in chunks of at
+// most buscalls.SeatChunkBytes on the call's chunk subjects, each acknowledged before the next (A-237, A-139). The input is of any
+// size. A caller that does not acknowledge within buscalls.SeatAckWithin ends the send, said in logf.
+func InputVerb(st Store, objs stateobjects.Objects, line ChunkLine, within time.Duration, logf func(string, ...any)) func(caller string, body json.RawMessage) (any, error) {
+	return func(_ string, raw json.RawMessage) (any, error) {
 		var in meshcatalog.InputInput
-		if raw, err := json.Marshal(args); err == nil {
-			_ = json.Unmarshal(raw, &in)
+		if err := json.Unmarshal(raw, &in); err != nil {
+			return nil, fmt.Errorf("the arguments are not JSON: %w", err)
 		}
-		return Input(context.Background(), st, in)
-	}}
+		if !callToken.MatchString(in.Call) {
+			return nil, errors.New("call is one token of 16 to 64 letters, digits, - or _: 128 random bits the caller chooses")
+		}
+		body, err := InputBody(context.Background(), st, objs, in.Digest)
+		if err != nil {
+			return nil, err
+		}
+		send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: in.Call}
+		total := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes)
+		acks := make(chan string, 1)
+		// heard before the answer goes, since the caller acknowledges only after it
+		stop, err := line.Hear(buscalls.SeatAcks(meshcatalog.Seat, meshcatalog.VerbInput), func(subject string) {
+			select {
+			case acks <- subject:
+			default:
+			}
+		})
+		if err != nil {
+			return nil, fmt.Errorf("the acknowledgements cannot be heard, so no chunk is sent: %w", err)
+		}
+		go func() {
+			defer stop()
+			for n := int64(1); n <= total; n++ {
+				from := (n - 1) * buscalls.SeatChunkBytes
+				to := min(int64(len(body)), from+buscalls.SeatChunkBytes)
+				data, _ := json.Marshal(buscalls.SeatChunk{N: n, Of: total, Data: body[from:to]})
+				if err := line.Publish(send.Chunk(n), data); err != nil {
+					logf("the send %s of the input %s ends at chunk %d of %d: %v", in.Call, in.Digest, n, total, err)
+					return
+				}
+				if !acknowledged(acks, send.Ack(n), within) {
+					logf("the send %s of the input %s ends at chunk %d of %d: no acknowledgement within %s", in.Call, in.Digest, n, total, within)
+					return
+				}
+			}
+		}()
+		return meshcatalog.InputAnswer{Digest: in.Digest, Chunks: total}, nil
+	}
 }
 
 // InputsAnswer is what indexer_inputs answers: each key kept with its time, digest and size, or one input's bytes.
@@ -200,7 +299,7 @@ type InputKept struct {
 
 // InputsReadTool is indexer_inputs (class read): every input the indexer keeps, with its key, digest and size; with a digest,
 // that input whole (base64).
-func InputsReadTool(st Store) stdio.Tool {
+func InputsReadTool(st Store, objs stateobjects.Objects) stdio.Tool {
 	return stdio.Tool{Name: InputsTool, Description: "Read the indexer's inputs of the ask run (design 47 G section 354): each key (write time and sha256) with its size; with a digest, that input's exact bytes, base64.",
 		Input: map[string]any{"type": "object", "properties": map[string]any{"digest": map[string]any{"type": "string", "description": "one input's sha256; none lists every key"}}},
 		Run: func(args map[string]any) (any, error) {
@@ -221,9 +320,18 @@ func InputsReadTool(st Store) stdio.Tool {
 				if err != nil || !found {
 					continue
 				}
-				text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw)
-				kept := InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))}
+				var kept InputKept
+				if ref, isRef := stateobjects.ReadRef(raw); isRef {
+					kept = InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(int(ref.Size) - 2)} // the quotes of the JSON string
+				} else {
+					text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw)
+					kept = InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))}
+				}
 				if want != "" {
+					var text string
+					if ok, err := inputValues(st, objs).Get(k, &text); err != nil || !ok {
+						continue
+					}
 					kept.Input = text
 				}
 				out.Inputs = append(out.Inputs, kept)
@@ -231,3 +339,19 @@ func InputsReadTool(st Store) stdio.Tool {
 			return out, nil
 		}}
 }
+
+// acknowledged waits within for the acknowledgement on subject, skipping another call's.
+func acknowledged(acks <-chan string, subject string, within time.Duration) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), within)
+	defer cancel()
+	for {
+		select {
+		case got := <-acks:
+			if got == subject {
+				return true
+			}
+		case <-ctx.Done():
+			return false
+		}
+	}
+}

modules/mesh-delivery-indexer/internal/index/versions_test.go

@@ -260,11 +260,10 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	a, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1})
-	if err != nil || a.Digest != d1 || a.Input != "Zmlyc3Q=" {
-		t.Fatalf("the input read back: %+v, %v", a, err)
+	if a, err := InputBody(ctx, ix.Inputs, nil, d1); err != nil || string(a) != "first" {
+		t.Fatalf("the input read back: %q, %v", a, err)
 	}
-	if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: "sha256:none"}); err == nil {
+	if _, err := InputBody(ctx, ix.Inputs, nil, "sha256:none"); err == nil {
 		t.Error("an input never kept is answered")
 	}
 	now = now.Add(23 * time.Hour)
@@ -292,10 +291,10 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
 	if all, _ := ix.Inputs.Keys(ctx); slices.Contains(all, InputsAskPrefix+"ask-1") {
 		t.Errorf("the name by ask of the swept input stays: %v", all)
 	}
-	if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil {
+	if _, err := InputBody(ctx, ix.Inputs, nil, d1); err == nil {
 		t.Error("the input a day old is still answered")
 	}
-	if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d3}); err != nil {
+	if _, err := InputBody(ctx, ix.Inputs, nil, d3); err != nil {
 		t.Error(err)
 	}
 }
@@ -517,16 +516,16 @@ func TestAnInputIsDeletedAtTheStepEndedOfItsAsk(t *testing.T) {
 	if err := ix.HearInputEnded(ctx, nodebuildgo.StepEnded{Ask: "a.1"}); err != nil {
 		t.Errorf("an end heard again: %v", err)
 	}
-	if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil {
+	if _, err := InputBody(ctx, ix.Inputs, nil, d1); err == nil {
 		t.Error("the input of an ended ask is still answered")
 	}
-	if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d2}); err != nil {
+	if _, err := InputBody(ctx, ix.Inputs, nil, d2); err != nil {
 		t.Errorf("the input of another ask: %v", err)
 	}
 	if got, _ := ix.Inputs.Keys(ctx); len(got) != 2 {
 		t.Errorf("keys after one end: %v", got)
 	}
-	if a, err := InputsReadTool(ix.Inputs).Run(map[string]any{}); err != nil || len(a.(InputsAnswer).Inputs) != 1 {
+	if a, err := InputsReadTool(ix.Inputs, nil).Run(map[string]any{}); err != nil || len(a.(InputsAnswer).Inputs) != 1 {
 		t.Errorf("indexer_inputs counts a key by ask: %+v, %v", a, err)
 	}
 }

modules/mesh-delivery-indexer/manifest_test.go

@@ -382,7 +382,9 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR
 			for _, f := range forbidden {
 				// The seat list reader's Bus interface names a method Subscribe; the indexer's own bus hands it its own events
 				// directly and subscribes to nothing on the bus.
-				if f == "Subscribe(" && strings.HasSuffix(path, "seatlist.go") {
+				// The acknowledgements of the chunks of the verb input are heard on the core connection while a send runs (design 47 G
+				// §712, A-237, §681): cmd/mesh-delivery-indexer/inputs.go.
+				if f == "Subscribe(" && (strings.HasSuffix(path, "seatlist.go") || strings.HasSuffix(path, "mesh-delivery-indexer/inputs.go")) {
 					continue
 				}
 				if strings.Contains(string(src), f) {

modules/mesh-delivery-sdk/buildstep/ask_test.go

@@ -113,7 +113,7 @@ func TestAskCheckAndRoundTrip(t *testing.T) {
 		{"input on a state", func(a *StepAsk) { a.Run.Input.Verb = "mesh-delivery-checker.inputs" }, "no seat:<seat>.<verb>"},
 		{"input on one machine", func(a *StepAsk) { a.Run.Input.Verb = "seat:mesh-delivery-checker.input@ace" }, "no seat:<seat>.<verb>"},
 		{"input by no digest", func(a *StepAsk) { a.Run.Input.Digest = "d1" }, "no sha256 digest"},
-		{"input over its bound", func(a *StepAsk) { a.Run.Input.Size = RunInputBound + 1 }, "outside 1 to"},
+		{"input of no bytes", func(a *StepAsk) { a.Run.Input.Size = 0 }, "under 1"},
 		{"argument of the shell", func(a *StepAsk) { a.Run.Args = []string{"-x; rm"} }, "no word or flag"},
 	} {
 		a := runAsk()
@@ -148,8 +148,9 @@ func TestAskCheckAndRoundTrip(t *testing.T) {
 	if a := stepAsk(); !a.Step.Skips("./modules/nats/a", "sha256:bbbb") || a.Step.Skips("./modules/nats/a", "sha256:cccc") {
 		t.Error("a kept result skips its package at its digest and level only")
 	}
-	if _, err := RunInputOf("seat:mesh-delivery-checker.input", make([]byte, RunInputBound+1)); err == nil || !strings.Contains(err.Error(), "262145 bytes") {
-		t.Errorf("an input over its bound: %v", err)
+	// A-237: a run's input is any size.
+	if in, err := RunInputOf("seat:mesh-delivery-checker.input", make([]byte, 3<<20)); err != nil || in.Size != 3<<20 {
+		t.Errorf("an input of 3 MiB: %+v, %v", in, err)
 	}
 	in := runAsk().Run.Input
 	if err := in.Verify([]byte(`{"v":1}`)); err != nil {

modules/mesh-delivery-sdk/buildstep/run.go

@@ -20,8 +20,8 @@ import (
 // Input names, checks its digest and writes it to the file RunInputEnv names; then runs the program with Args, no
 // network and no bus account (ADR 0448 §2, §4), and answers RunAnswer on its step key.
 //
-// The input is carried by reference, as a compose carries its manifests: the asker keeps it by its digest, measured
-// against RunInputBound before it asks, and serves it on a verb of its seat (RunInput.Verb); the build module asks that
+// The input is carried by reference, as a compose carries its manifests: the asker keeps it by its digest, of any
+// size (A-237), and serves it on a verb of its seat (RunInput.Verb); the build module asks that
 // verb as its manifest's `invokes` grants it, never reading the asker's state (ADR 0596 §2), so the ask on the stream
 // never carries it (issue 1930). The layouts are the part build-step's `RunJob`, `RunProgram`, `RunHead`, `RunInput`
 // and `RunAnswer`.
@@ -77,22 +77,14 @@ type RunAnswer struct {
 	Said   string          `json:"said,omitempty"`
 }
 
-// RunInputBound is the most bytes a run's input may hold: the bus's value limit the design already uses (256 KiB), so
-// the asker keeps it as one value (the lead's ruling on milestone 47/B0, 2026-11-06).
-const RunInputBound = 256 << 10
-
 // RunHeadsEnv names, for the program, the directory its heads are laid in; RunInputEnv the file its input is written to.
 const (
 	RunHeadsEnv = "MESH_RUN_HEADS"
 	RunInputEnv = "MESH_RUN_INPUT"
 )
 
-// RunInputOf is the reference to body served on verb (`seat:<seat>.<verb>`), refusing a body over RunInputBound, naming
-// its size, so an asker refuses it before asking.
+// RunInputOf is the reference to body served on verb (`seat:<seat>.<verb>`); a body is of any size.
 func RunInputOf(verb string, body []byte) (RunInput, error) {
-	if len(body) > RunInputBound {
-		return RunInput{}, fmt.Errorf("buildstep: a run's input of %d bytes is over its bound of %d bytes", len(body), RunInputBound)
-	}
 	sum := sha256.Sum256(body)
 	in := RunInput{Verb: verb, Digest: hex.EncodeToString(sum[:]), Size: len(body)}
 	return in, in.check("")
@@ -118,8 +110,8 @@ func (i RunInput) check(id string) error {
 		return fmt.Errorf("buildstep: the run ask %s serves its input on %q, which is no seat:<seat>.<verb> of no one machine", id, i.Verb)
 	case !runDigestOK.MatchString(i.Digest):
 		return fmt.Errorf("buildstep: the run ask %s names its input by %q, which is no sha256 digest", id, i.Digest)
-	case i.Size <= 0 || i.Size > RunInputBound:
-		return fmt.Errorf("buildstep: the run ask %s names an input of %d bytes, outside 1 to %d", id, i.Size, RunInputBound)
+	case i.Size <= 0:
+		return fmt.Errorf("buildstep: the run ask %s names an input of %d bytes, under 1", id, i.Size)
 	}
 	return nil
 }