From 842a901229263743a87784333029e2d47b330822 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 16:48:22 +0100 Subject: [PATCH] A-291: a process receives a seat verb's chunked answer through one helper Conn.CallChunks subscribes to the call's chunks before it asks the verb, hands the verb's answer to an accept hook, then receives the chunks the answer counts, in order, acknowledging each, into a writer; CallFile answers the whole body as the member file, the form a bundle's seat call answers. The checker, the indexer, the assigner and the recovery keeper read a step's answer kept as an object through it. Builds: design 47 G A-291, A-290, A-254 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/chunks.go | 110 ++++++++++++++++++++++++++ go/process/chunks_integration_test.go | 71 +++++++++++++++++ 2 files changed, 181 insertions(+) create mode 100644 go/process/chunks.go create mode 100644 go/process/chunks_integration_test.go diff --git a/go/process/chunks.go b/go/process/chunks.go new file mode 100644 index 00000000..ee5c14e9 --- /dev/null +++ b/go/process/chunks.go @@ -0,0 +1,110 @@ +package process + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/calladdress" +) + +// CallChunks calls a seat verb whose answer file is sent in chunks (a verb its seat definition marks `chunks`, design 47 G +// ยง681, A-139), the one way a process receives such an answer (A-291): it subscribes to the chunks of call (the 128 random +// bits the verb's input carries as its call) before it asks, so none is missed, asks key ("seat:.[@]") +// with body as Call does, hands the verb's answer to accept (nil accepts every answer; an error ends the call before a chunk +// is taken), then receives the number of chunks the answer's member `chunks` names, in order, acknowledging each, into w. +// It answers the verb's answer. A chunk that does not come within buscalls.SeatAckWithin, or ctx ending, ends the receive. +func (c *Conn) CallChunks(ctx context.Context, key, call string, body any, accept func(answer json.RawMessage) error, w io.Writer) (json.RawMessage, error) { + a, err := calladdress.Parse(key) + if err != nil { + return nil, err + } + if a.Seat == "" { + return nil, fmt.Errorf("%q names no seat verb, so it sends no chunks", key) + } + if call == "" { + return nil, fmt.Errorf("the call of %s names no call token for its chunks", key) + } + send := buscalls.SeatChunks{Seat: a.Seat, Verb: a.Verb, Call: call} + chunks := make(chan *nats.Msg, 4) + sub, err := c.nc.ChanSubscribe(send.Chunks(), chunks) + if err != nil { + return nil, fmt.Errorf("the chunks of %s cannot be heard: %w", key, err) + } + defer func() { _ = sub.Unsubscribe() }() + answer, err := c.Call(key, body) + if err != nil { + return nil, err + } + if accept != nil { + if err := accept(answer); err != nil { + return answer, err + } + } + var count struct { + Chunks int64 `json:"chunks"` + } + if err := json.Unmarshal(answer, &count); err != nil { + return answer, fmt.Errorf("%s answered no count of its chunks: %w", key, err) + } + next := func() (string, []byte, error) { + bound := time.NewTimer(buscalls.SeatAckWithin) + defer bound.Stop() + select { + case m := <-chunks: + return m.Subject, m.Data, nil + case <-bound.C: + return "", nil, buscalls.ErrSeatChunkMissed + case <-ctx.Done(): + return "", nil, ctx.Err() + } + } + ack := func(subject string) error { + _, err := c.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin) + return err + } + if _, err := send.Receive(count.Chunks, next, ack, w); err != nil { + return answer, fmt.Errorf("the answer of %s: %w", key, err) + } + return answer, nil +} + +// CallFile is CallChunks for an answer read whole, in the form a bundle's ordinary seat call answers a chunked verb (design 47 +// G A-249, A-253): the verb's answer with the whole file added as its member `file` (base64), none when no chunk was sent. The +// call token is body's member `call`. +func (c *Conn) CallFile(key string, body any) (json.RawMessage, error) { + call := callOf(body) + var file bytes.Buffer + answer, err := c.CallChunks(context.Background(), key, call, body, nil, &file) + if err != nil { + return nil, err + } + var whole map[string]any + if err := json.Unmarshal(answer, &whole); err != nil { + return nil, fmt.Errorf("%s answered what is no object: %w", key, err) + } + if file.Len() > 0 { + whole["file"] = base64.StdEncoding.EncodeToString(file.Bytes()) + } + return json.Marshal(whole) +} + +// callOf is the member call of a verb's input. +func callOf(body any) string { + raw, err := json.Marshal(body) + if err != nil { + return "" + } + var in struct { + Call string `json:"call"` + } + _ = json.Unmarshal(raw, &in) + return in.Call +} diff --git a/go/process/chunks_integration_test.go b/go/process/chunks_integration_test.go new file mode 100644 index 00000000..60d83d67 --- /dev/null +++ b/go/process/chunks_integration_test.go @@ -0,0 +1,71 @@ +//go:build integration + +package process + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// Design 47 G A-291: a process receives a seat verb's chunked answer through one helper: it subscribes to the call's chunks +// before it asks, acknowledges each, and answers the whole body (CallChunks into a writer, CallFile as the member file); +// an answer accept refuses takes no chunk. +func TestAProcessReceivesAChunkedSeatAnswerWhole(t *testing.T) { + url := testbus.URL(t) + holder, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + body := []byte(strings.Repeat("0123456789", 2*buscalls.SeatChunkBytes/10+7)) // three chunks + of := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes) + sub, err := holder.Subscribe("mesh.seat.node-build-go.call.output.ace.>", func(m *nats.Msg) { + var in struct{ Ask, Call string } + _ = json.Unmarshal(m.Data, &in) + answer, _ := json.Marshal(map[string]any{"result": map[string]any{"ask": in.Ask, "chunks": of}}) + _ = m.Respond(answer) + go func() { + _ = buscalls.SeatChunks{Seat: "node-build-go", Verb: "output", Call: in.Call}.Send(holder, body, buscalls.SeatAckWithin, t.Logf) + }() + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + _ = holder.Flush() + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + c := &Conn{nc: nc, account: Account{User: "novox.mesh-delivery-checker"}} + defer c.Close() + var got bytes.Buffer + answer, err := c.CallChunks(context.Background(), "seat:node-build-go.output@ace", "c1", map[string]any{"ask": "a1", "call": "c1"}, nil, &got) + if err != nil || !bytes.Equal(got.Bytes(), body) || !strings.Contains(string(answer), `"ask":"a1"`) { + t.Fatalf("answered %s and %d bytes (%v), want the verb's answer and the whole %d bytes", answer, got.Len(), err, len(body)) + } + raw, err := c.CallFile("seat:node-build-go.output@ace", map[string]any{"ask": "a2", "call": "c2"}) + var file struct{ Ask, File string } + if err != nil || json.Unmarshal(raw, &file) != nil || file.Ask != "a2" { + t.Fatalf("CallFile answered %.80s, %v", raw, err) + } + if whole, _ := base64.StdEncoding.DecodeString(file.File); !bytes.Equal(whole, body) { + t.Fatalf("CallFile's file holds %d bytes, want %d", len(whole), len(body)) + } + refused := errors.New("too large for this disk") + got.Reset() + if _, err := c.CallChunks(context.Background(), "seat:node-build-go.output@ace", "c3", map[string]any{"ask": "a3", "call": "c3"}, + func(json.RawMessage) error { return refused }, &got); !errors.Is(err, refused) || got.Len() != 0 { + t.Fatalf("an answer accept refused: %v, %d bytes taken", err, got.Len()) + } +}