B5g3-5: the tool runner receives a chunked seat answer and hands the bundle the whole

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

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

node-tools/internal/bus/chunked.go

@@ -0,0 +1,95 @@
+package bus
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// FileField is the member of a chunked verb's answer that carries the file received: its whole bytes, base64 (design 47 G
+// A-249).
+const FileField = "file"
+
+// AskAsChunked is AskAs of a seat's verb whose definition says `chunks` (design 47 G §681, A-237, A-249): it chooses the call
+// token (the body's `call`), hears the chunk subjects of that call before it asks, receives the chunks the verb's answer
+// counts with buscalls.SeatChunks, each acknowledged before the next, and answers the verb's own answer with the whole
+// file added as FileField. The asking bundle receives no chunk itself.
+func (c *Conn) AskAsChunked(module, key string, body any) (Answered, error) {
+	seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
+	if !strings.HasPrefix(key, "seat:") || !ok || strings.Contains(verb, "@") {
+		return Answered{}, fmt.Errorf("%q is no seat:<seat>.<verb> of no one machine, which a chunked answer needs", key)
+	}
+	args := map[string]any{}
+	if raw, err := json.Marshal(body); err != nil {
+		return Answered{}, err
+	} else if string(raw) != "null" {
+		if err := json.Unmarshal(raw, &args); err != nil {
+			return Answered{}, fmt.Errorf("the arguments of %s are not an object: %w", key, err)
+		}
+	}
+	var bits [16]byte
+	if _, err := rand.Read(bits[:]); err != nil {
+		return Answered{}, err
+	}
+	send := buscalls.SeatChunks{Seat: seat, Verb: verb, Call: hex.EncodeToString(bits[:])}
+	args["call"] = send.Call
+	chunks := make(chan *nats.Msg, 4)
+	sub, err := c.nc.ChanSubscribe(send.Chunks(), chunks)
+	if err != nil {
+		return Answered{}, fmt.Errorf("the chunks of %s cannot be heard: %w", key, err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	answered, err := c.AskAs(module, key, args)
+	if err != nil {
+		return answered, err
+	}
+	fields := map[string]json.RawMessage{}
+	if err := json.Unmarshal(answered.Result, &fields); err != nil {
+		return answered, fmt.Errorf("%s answered what is no object: %w", key, err)
+	}
+	var count int64
+	if raw, has := fields["chunks"]; has {
+		if err := json.Unmarshal(raw, &count); err != nil {
+			return answered, fmt.Errorf("%s answered chunks that is no count: %w", key, err)
+		}
+	}
+	var file bytes.Buffer
+	if count > 0 {
+		next := func() (string, []byte, error) {
+			select {
+			case m := <-chunks:
+				return m.Subject, m.Data, nil
+			case <-time.After(buscalls.SeatAckWithin):
+				return "", nil, buscalls.ErrSeatChunkMissed
+			}
+		}
+		ack := func(subject string) error {
+			_, err := c.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin)
+			return err
+		}
+		if _, err := send.Receive(count, next, ack, &file); err != nil {
+			return answered, fmt.Errorf("the chunks of %s: %w", key, err)
+		}
+	}
+	if _, taken := fields[FileField]; taken {
+		return answered, errors.New(key + " answers a member named " + FileField + " itself")
+	}
+	enc, _ := json.Marshal(base64.StdEncoding.EncodeToString(file.Bytes()))
+	fields[FileField] = enc
+	out, err := json.Marshal(fields)
+	if err != nil {
+		return answered, err
+	}
+	answered.Result = out
+	return answered, nil
+}

node-tools/internal/bus/chunked_test.go

@@ -0,0 +1,75 @@
+package bus_test
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
+)
+
+// Design 47 G A-249, §681: the runner asks a chunked verb with a call token of its own, receives the chunks the answer counts
+// (each acknowledged before the next) and answers the verb's own answer with the whole file, so the asker receives none.
+func TestAChunkedSeatAnswerIsReceivedByTheRunnerAndHandedWhole(t *testing.T) {
+	url := mt.URL(t)
+	holder := dial(t, url, "mesh-delivery-indexer", "anchor", "")
+	asker := dial(t, url, "mesh-build-go", "anchor", "")
+	raw, err := nats.Connect(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(raw.Close)
+	file := make([]byte, buscalls.SeatChunkBytes*2+5)
+	for i := range file {
+		file[i] = byte(i)
+	}
+	// the holder's process: answers the count, then sends each chunk and waits for its acknowledgement
+	stop, err := holder.Handle("seat:mesh-catalog.input", func(body json.RawMessage) (any, error) {
+		var in struct{ Call, Digest string }
+		if err := json.Unmarshal(body, &in); err != nil || in.Call == "" {
+			t.Errorf("the verb was asked with no call token: %s (%v)", body, err)
+			return nil, err
+		}
+		send := buscalls.SeatChunks{Seat: "mesh-catalog", Verb: "input", Call: in.Call}
+		go func() {
+			for n := int64(1); n <= 3; n++ {
+				end := min(int64(len(file)), n*buscalls.SeatChunkBytes)
+				data, _ := json.Marshal(buscalls.SeatChunk{N: n, Of: 3, Data: file[(n-1)*buscalls.SeatChunkBytes : end]})
+				ackSub, _ := raw.SubscribeSync(send.Ack(n))
+				_ = raw.Flush()
+				if err := raw.Publish(send.Chunk(n), data); err != nil {
+					return
+				}
+				m, err := ackSub.NextMsg(buscalls.SeatAckWithin)
+				if err != nil {
+					t.Errorf("chunk %d not acknowledged: %v", n, err)
+					return
+				}
+				_ = m.Respond([]byte("{}"))
+			}
+		}()
+		return map[string]any{"digest": in.Digest, "chunks": 3}, nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	got, err := asker.AskAsChunked("mesh-build-go", "seat:mesh-catalog.input", map[string]any{"digest": "d1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var ans struct {
+		Digest string `json:"digest"`
+		Chunks int    `json:"chunks"`
+		File   string `json:"file"`
+	}
+	if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Digest != "d1" || ans.Chunks != 3 {
+		t.Fatalf("the answer %s: %v", got.Result, err)
+	}
+	if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) {
+		t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err)
+	}
+}

node-tools/internal/runtime/chunked_test.go

@@ -0,0 +1,29 @@
+package runtime
+
+import (
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
+)
+
+// Design 47 G A-249: a bundle's seat call is received by the runner in chunks only for a verb whose definition says `chunks`,
+// of no one machine.
+func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) {
+	c := &consumers{definitions: func() map[string]seatdef.Definition {
+		return map[string]seatdef.Definition{"mesh-catalog": {Verbs: []seatdef.Verb{{Name: "input", Chunks: true}, {Name: "entry"}}}}
+	}}
+	for key, want := range map[string]bool{
+		"seat:mesh-catalog.input":       true,
+		"seat:mesh-catalog.entry":       false,
+		"seat:mesh-catalog.input@ace":   false,
+		"seat:mesh-unread.input":        false,
+		"mesh-delivery-indexer.indexer": false,
+	} {
+		if got := c.chunked(key); got != want {
+			t.Errorf("chunked(%s) = %v, want %v", key, got, want)
+		}
+	}
+	if (&consumers{}).chunked("seat:mesh-catalog.input") {
+		t.Error("a runner that holds no seat definitions asks no chunks")
+	}
+}

node-tools/internal/runtime/gate.go

@@ -2,6 +2,7 @@ package runtime
 
 import (
 	"fmt"
+	"strings"
 
 	"git.novox.be/novox/mesh-sdk/go/approval"
 	"git.novox.be/novox/mesh-sdk/go/seatdef"
@@ -86,3 +87,21 @@ func (r *runner) unservedMarks(module string, regs []launch.Registration) []stri
 	}
 	return r.events.gate.Unserved(module, r.fingerprintOf(module), registered)
 }
+
+// chunked says whether key names a seat's verb whose definition says `chunks` (design 47 G A-249): its answer is received in
+// chunks by the runner, whose call hands the bundle the whole.
+func (c *consumers) chunked(key string) bool {
+	if c.definitions == nil || !strings.HasPrefix(key, "seat:") {
+		return false
+	}
+	seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
+	if !ok || strings.Contains(verb, "@") {
+		return false
+	}
+	for _, v := range c.definitions()[seat].Verbs {
+		if v.Name == verb {
+			return v.Chunks
+		}
+	}
+	return false
+}

node-tools/internal/runtime/runtime.go

@@ -915,7 +915,11 @@ func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) {
 	if len(asked.Body) == 0 {
 		body = map[string]any{}
 	}
-	answered, err := b.all.conn.AskAs(b.module, asked.Key, body)
+	ask := b.all.conn.AskAs
+	if b.all.chunked(asked.Key) {
+		ask = b.all.conn.AskAsChunked // the runner receives the chunks of the answer; the bundle receives none (design 47 G A-249)
+	}
+	answered, err := ask(b.module, asked.Key, body)
 	if err != nil {
 		return nil, err
 	}