diff --git a/node-tools/internal/bus/chunked.go b/node-tools/internal/bus/chunked.go index d1da74c..7b9257b 100644 --- a/node-tools/internal/bus/chunked.go +++ b/node-tools/internal/bus/chunked.go @@ -25,9 +25,12 @@ const FileField = "file" // 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:. of no one machine, which a chunked answer needs", key) + // The verb may be asked of the machine that ran the ask (`@`, A-254): the chunk subjects name the seat, the verb and + // the call alone. + name, _, _ := strings.Cut(key, "@") + seat, verb, ok := strings.Cut(strings.TrimPrefix(name, "seat:"), ".") + if !strings.HasPrefix(key, "seat:") || !ok { + return Answered{}, fmt.Errorf("%q is no seat:., which a chunked answer needs", key) } args := map[string]any{} if raw, err := json.Marshal(body); err != nil { @@ -93,3 +96,48 @@ func (c *Conn) AskAsChunked(module, key string, body any) (Answered, error) { answered.Result = out return answered, nil } + +// AnswerChunked is a seat verb's answer as its definition says `chunks` (design 47 G A-257, ยง681): a bundle answers the whole file +// as the member FileField (base64), and the runner that serves the verb sends it in chunks with buscalls.SeatChunks on the call +// token the body names, each acknowledged by the caller before the next, and answers the verb's own members with `chunks`, the +// count, in place of the file. The bundle sends no chunk itself. An answer with no file is answered as it is. The send runs after +// the answer is returned: the caller hears the chunks before it asks (AskAsChunked) and acknowledges each once it has the answer. +func (c *Conn) AnswerChunked(seat, verb string, body json.RawMessage, answer any) (any, error) { + raw, err := json.Marshal(answer) + if err != nil { + return nil, err + } + fields := map[string]json.RawMessage{} + if err := json.Unmarshal(raw, &fields); err != nil { + return answer, nil + } + enc, has := fields[FileField] + if !has { + return answer, nil + } + var text string + if err := json.Unmarshal(enc, &text); err != nil { + return nil, fmt.Errorf("%s.%s answers a %s that is no string: %w", seat, verb, FileField, err) + } + file, err := base64.StdEncoding.DecodeString(text) + if err != nil { + return nil, fmt.Errorf("%s.%s answers a %s in no base64: %w", seat, verb, FileField, err) + } + var asked struct { + Call string `json:"call"` + } + if err := json.Unmarshal(body, &asked); err != nil || asked.Call == "" { + return nil, fmt.Errorf("%s.%s is asked in chunks with no call token", seat, verb) + } + delete(fields, FileField) + count := int64((len(file) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes) + fields["chunks"], _ = json.Marshal(count) + send := buscalls.SeatChunks{Seat: seat, Verb: verb, Call: asked.Call} + go func() { _ = send.Send(c.nc, file, buscalls.SeatAckWithin, c.Logf) }() + return json.RawMessage(mustMarshal(fields)), nil +} + +func mustMarshal(v any) []byte { + b, _ := json.Marshal(v) + return b +} diff --git a/node-tools/internal/bus/chunked_test.go b/node-tools/internal/bus/chunked_test.go index a14741e..5e0d74c 100644 --- a/node-tools/internal/bus/chunked_test.go +++ b/node-tools/internal/bus/chunked_test.go @@ -73,3 +73,44 @@ func TestAChunkedSeatAnswerIsReceivedByTheRunnerAndHandedWhole(t *testing.T) { t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err) } } + +// Design 47 G A-257, A-254: a bundle answers a chunked verb with the whole file in the member `file`; the runner that serves the +// verb sends the file in chunks (each acknowledged before the next) and answers the count, so the bundle sends none, and the asker +// of one node's verb receives it whole. +func TestABundlesChunkedAnswerIsSentByTheRunner(t *testing.T) { + url := mt.URL(t) + holder := dial(t, url, "mesh-build-go", "anchor", "") + asker := dial(t, url, "mesh-delivery-indexer", "anchor", "") + file := make([]byte, buscalls.SeatChunkBytes*2+7) + for i := range file { + file[i] = byte(i * 7) + } + stop, err := holder.Handle("seat:node-build-go.output@anchor", func(body json.RawMessage) (any, error) { + answer := map[string]any{"ask": "a1", "file": base64.StdEncoding.EncodeToString(file)} + return holder.AnswerChunked("node-build-go", "output", body, answer) + }) + if err != nil { + t.Fatal(err) + } + defer stop() + got, err := asker.AskAsChunked("mesh-delivery-indexer", "seat:node-build-go.output@anchor", map[string]any{"ask": "a1"}) + if err != nil { + t.Fatal(err) + } + var ans struct { + Ask string `json:"ask"` + Chunks int `json:"chunks"` + File string `json:"file"` + } + if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Ask != "a1" || 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) + } + // an answer with no file is the verb's own: no chunks are sent + plain, err := holder.AnswerChunked("node-build-go", "output", json.RawMessage(`{"call":"c"}`), map[string]any{"ask": "x", "chunks": 0}) + if m, _ := plain.(map[string]any); err != nil || m["chunks"] != 0 { + t.Fatalf("an answer with no file was %v, %v", plain, err) + } +} diff --git a/node-tools/internal/runtime/chunked_test.go b/node-tools/internal/runtime/chunked_test.go index 3b929c1..34da586 100644 --- a/node-tools/internal/runtime/chunked_test.go +++ b/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, } { diff --git a/node-tools/internal/runtime/gate.go b/node-tools/internal/runtime/gate.go index 960f299..02a605f 100644 --- a/node-tools/internal/runtime/gate.go +++ b/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 { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 10dfa64..ebeeec9 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/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)