diff --git a/node-tools/internal/bus/chunked.go b/node-tools/internal/bus/chunked.go new file mode 100644 index 0000000..d1da74c --- /dev/null +++ b/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:. 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 +} diff --git a/node-tools/internal/bus/chunked_test.go b/node-tools/internal/bus/chunked_test.go new file mode 100644 index 0000000..a14741e --- /dev/null +++ b/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) + } +} diff --git a/node-tools/internal/runtime/chunked_test.go b/node-tools/internal/runtime/chunked_test.go new file mode 100644 index 0000000..3b929c1 --- /dev/null +++ b/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") + } +} diff --git a/node-tools/internal/runtime/gate.go b/node-tools/internal/runtime/gate.go index 61af8a6..960f299 100644 --- a/node-tools/internal/runtime/gate.go +++ b/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 +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index d42b3a4..c391053 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/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 }