Merge branch 'feat/47-G-fix11-c1c' into feat/47-G-join11

node-tools/internal/runtime/chunked_test.go

@@ -7,7 +7,7 @@ import (
 )
 
 // 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.
+// of any machine: a verb a node holds is asked of the node that ran the ask (A-254).
 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"}}}}
@@ -15,7 +15,8 @@ func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) {
 	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-catalog.input@ace":   true,
+		"seat:mesh-catalog.entry@ace":   false,
 		"seat:mesh-unread.input":        false,
 		"mesh-delivery-indexer.indexer": false,
 	} {

node-tools/internal/runtime/gate.go

@@ -94,8 +94,9 @@ 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, "@") {
+	name, _, _ := strings.Cut(key, "@") // a verb asked of the machine that ran the ask is chunked as the verb is (A-254)
+	seat, verb, ok := strings.Cut(strings.TrimPrefix(name, "seat:"), ".")
+	if !ok {
 		return false
 	}
 	for _, v := range c.definitions()[seat].Verbs {

node-tools/internal/runtime/runtime.go

@@ -310,7 +310,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
 	}
 	logf("%s", line)
 	conn.OnMembership(func(bus.Membership) { go func() { r.refreshRepairs(); conn.Flush() }() })
-	stopSeats, again := serveSeats(conn, r.modules, r.registrations, r.events.gated, r.events.admitted, r.fingerprintOf, logf)
+	stopSeats, again := serveSeats(conn, r.modules, r.registrations, r.events.gated, r.events.admitted, r.events.chunked, r.fingerprintOf, logf)
 	r.seatsAgain = again
 	r.stops = append(r.stops, stopSeats)
 
@@ -717,7 +717,7 @@ func subjectsOf(conn *bus.Conn, module, tool string) []string {
 // whenever a membership changes, and when again is called because a bundle was loaded again. Whether
 // this machine holds the seat is the bus's to decide. It answers its stop and again.
 func serveSeats(conn *bus.Conn, modules []string, current func() []registration, gated gatedRun,
-	admitted func(seat, verb, caller string) error, fingerprintOf func(string) string, logf func(string, ...any)) (func(), func()) {
+	admitted func(seat, verb, caller string) error, chunked func(key string) bool, fingerprintOf func(string) string, logf func(string, ...any)) (func(), func()) {
 	var mu sync.Mutex
 	var stops []func()
 	stopped := false
@@ -763,8 +763,13 @@ func serveSeats(conn *bus.Conn, modules []string, current func() []registration,
 					if err := admitted(v.Seat, v.Verb, caller); err != nil {
 						return nil, err
 					}
-					return gated(approval.Call{Address: v.Seat + "." + v.Verb, Module: module, Seat: v.Seat, Name: v.Verb,
+					answer, err := gated(approval.Call{Address: v.Seat + "." + v.Verb, Module: module, Seat: v.Seat, Name: v.Verb,
 						Fingerprint: fingerprintOf(module), Caller: caller, Args: argsOf(body), Description: t.Description}, t, caller)
+					if err != nil || !chunked("seat:"+v.Seat+"."+v.Verb) {
+						return answer, err
+					}
+					// A verb marked chunks: the bundle answers the whole file, the runner sends it in chunks (design 47 G A-257).
+					return conn.AnswerChunked(v.Seat, v.Verb, body, answer)
 				})
 				if err != nil {
 					logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)