B5g3-3: the indexer keeps an input of any size and answers input in chunks

Builds: design 47 G §712, §354, §681, A-237, A-139

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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) {