From 236fdc3a4a358a5b14a2ecfe03aac4807b355f84 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:25:42 +0100 Subject: [PATCH] B5g5-1: buscalls SeatChunks.Send, the holder's side of Receive; the mesh-catalog seat copies carry input in chunks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A holder sends a verb's answer file with SeatChunks.Send: it subscribes to the verb's acknowledgements, answers each as a request and publishes each chunk only after the one before is acknowledged. The bootstrap, seatlisttest and captured copies of the mesh-catalog seat are the contracts' file at the input verb with chunks. Builds: design 47 G §712, §681, A-237, A-139, A-249 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/buscalls/seatchunks.go | 68 ++++ go/buscalls/seatsend_test.go | 94 +++++ go/seatlist/bootstrap/mesh-catalog.json | 322 ++++++++++++++++-- .../seatlisttest/seats/mesh-catalog.json | 322 ++++++++++++++++-- .../seats/mesh-catalog/seat.json.captured | 322 ++++++++++++++++-- 5 files changed, 1065 insertions(+), 63 deletions(-) create mode 100644 go/buscalls/seatsend_test.go diff --git a/go/buscalls/seatchunks.go b/go/buscalls/seatchunks.go index 94a2648c..eaa09a25 100644 --- a/go/buscalls/seatchunks.go +++ b/go/buscalls/seatchunks.go @@ -7,6 +7,8 @@ import ( "io" "strconv" "time" + + "github.com/nats-io/nats.go" ) // SeatChunks is the send of one seat verb's answer file in chunks (design 47 G §681, A-64, A-139), a verb whose seat @@ -85,3 +87,69 @@ func (c SeatChunks) Receive(of int64, next func() (subject string, body []byte, } return written, nil } + +// ChunkLine is the core connection as a send of chunks reaches it, which a *nats.Conn is: a chunk is published on its subject and +// kept on no stream, and the acknowledgements are subscribed to, since the caller asks each as a request. +type ChunkLine interface { + Publish(subject string, data []byte) error + Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) +} + +// Send sends body in chunks of at most SeatChunkBytes, the holder's side of Receive (design 47 G §681, A-139, A-237): it subscribes +// to the verb's acknowledgements (SeatAcks), answers each as the request it is, publishes the nth chunk on Chunk(n) only once the +// n-1th is acknowledged on Ack(n-1), and ends the subscription when the send ends. An acknowledgement that does not come within +// `within` (SeatAckWithin where the caller has no other bound) ends the send with ErrSeatChunkMissed, and logf says where. Another +// call's acknowledgements are skipped. It blocks until the send ends, so a verb answers its chunk count first and runs it after. +func (c SeatChunks) Send(line ChunkLine, body []byte, within time.Duration, logf func(string, ...any)) error { + if logf == nil { + logf = func(string, ...any) {} + } + acks := make(chan string, 1) + sub, err := line.Subscribe(SeatAcks(c.Seat, c.Verb), func(m *nats.Msg) { + if m.Reply != "" { + _ = m.Respond([]byte("{}")) + } + select { + case acks <- m.Subject: + default: + } + }) + if err != nil { + return fmt.Errorf("the acknowledgements cannot be heard, so no chunk is sent: %w", err) + } + defer func() { _ = sub.Unsubscribe() }() + total := int64((len(body) + SeatChunkBytes - 1) / SeatChunkBytes) + for n := int64(1); n <= total; n++ { + from := (n - 1) * SeatChunkBytes + to := min(int64(len(body)), from+SeatChunkBytes) + data, err := json.Marshal(SeatChunk{N: n, Of: total, Data: body[from:to]}) + if err != nil { + return err + } + if err := line.Publish(c.Chunk(n), data); err != nil { + logf("the send %s of %s ends at chunk %d of %d: %v", c.Call, c.Verb, n, total, err) + return fmt.Errorf("chunk %d of %d: %w", n, total, err) + } + if !c.acknowledged(acks, c.Ack(n), within) { + logf("the send %s of %s ends at chunk %d of %d: no acknowledgement within %s", c.Call, c.Verb, n, total, within) + return fmt.Errorf("chunk %d of %d: %w", n, total, ErrSeatChunkMissed) + } + } + return nil +} + +// acknowledged waits within for the acknowledgement on subject, skipping another call's. +func (c SeatChunks) acknowledged(acks <-chan string, subject string, within time.Duration) bool { + timer := time.NewTimer(within) + defer timer.Stop() + for { + select { + case got := <-acks: + if got == subject { + return true + } + case <-timer.C: + return false + } + } +} diff --git a/go/buscalls/seatsend_test.go b/go/buscalls/seatsend_test.go new file mode 100644 index 00000000..6a81ecb6 --- /dev/null +++ b/go/buscalls/seatsend_test.go @@ -0,0 +1,94 @@ +package buscalls + +import ( + "bytes" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// lineMem is a core connection in memory: what Send publishes is kept, and when ackOf says so the chunk is acknowledged as the +// caller does, with a request on the ack subject, whose reply Send must give. +type lineMem struct { + mu sync.Mutex + handler nats.MsgHandler + subject string + chunks map[string][]byte + order []string + ackOf func(n int64) (string, bool) +} + +func (l *lineMem) Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.handler, l.subject = cb, subject + return nil, nil +} + +func (l *lineMem) Publish(subject string, data []byte) error { + l.mu.Lock() + l.chunks[subject] = data + l.order = append(l.order, subject) + h := l.handler + l.mu.Unlock() + var c SeatChunk + _ = json.Unmarshal(data, &c) + if sub, ok := l.ackOf(c.N); ok { + go h(&nats.Msg{Subject: sub, Reply: "_INBOX.x", Sub: nil}) + } + return nil +} + +// TestSeatChunksSend: the bytes go in chunks of at most SeatChunkBytes, each published only after the one before is +// acknowledged, and the receive of the same form reads them back; an acknowledgement missed or of another call ends the +// send (design 47 G §681, A-237, A-139). +func TestSeatChunksSend(t *testing.T) { + c := SeatChunks{Seat: "mesh-catalog", Verb: "input", Call: "0123456789abcdef"} + body := bytes.Repeat([]byte("x"), 2*SeatChunkBytes+5) + for _, tc := range []struct { + name string + ackOf func(int64) (string, bool) + err string + sent int + }{ + {name: "every chunk acknowledged", ackOf: func(n int64) (string, bool) { return c.Ack(n), true }, sent: 3}, + {name: "no acknowledgement", ackOf: func(int64) (string, bool) { return "", false }, err: "chunk 1 of 3", sent: 1}, + {name: "another call's acknowledgement", ackOf: func(n int64) (string, bool) { + return SeatChunks{Seat: "mesh-catalog", Verb: "input", Call: "fedcba9876543210"}.Ack(n), true + }, err: "chunk 1 of 3", sent: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + l := &lineMem{chunks: map[string][]byte{}, ackOf: tc.ackOf} + err := c.Send(l, body, time.Millisecond*50, func(string, ...any) {}) + if tc.err == "" && err != nil || tc.err != "" && (err == nil || !strings.Contains(err.Error(), tc.err)) { + t.Fatalf("err %v, want %q", err, tc.err) + } + if len(l.order) != tc.sent { + t.Fatalf("%d chunks sent, want %d", len(l.order), tc.sent) + } + if l.subject != SeatAcks("mesh-catalog", "input") { + t.Fatalf("acknowledgements heard on %q", l.subject) + } + if tc.err != "" { + return + } + var got bytes.Buffer + i := 0 + _, err = c.Receive(3, func() (string, []byte, error) { + s := l.order[i] + i++ + return s, l.chunks[s], nil + }, func(string) error { return nil }, &got) + if err != nil || !bytes.Equal(got.Bytes(), body) { + t.Fatalf("received %d bytes, %v", got.Len(), err) + } + }) + } + if err := c.Send(&lineMem{chunks: map[string][]byte{}}, nil, time.Second, nil); err != nil { + t.Errorf("no bytes: %v", err) + } +} diff --git a/go/seatlist/bootstrap/mesh-catalog.json b/go/seatlist/bootstrap/mesh-catalog.json index 8b986d9b..8c9841db 100644 --- a/go/seatlist/bootstrap/mesh-catalog.json +++ b/go/seatlist/bootstrap/mesh-catalog.json @@ -171,7 +171,7 @@ }, { "name": "seat-definitions", - "description": "The seat list (ADR 0636 \u00a74): with no seat, every trunk seat with its digest and the revision of the list; with a seat, that trunk definition resolved with the parts it includes. Reads only.", + "description": "The seat list (ADR 0636 §4): with no seat, every trunk seat with its digest and the revision of the list; with a seat, that trunk definition resolved with the parts it includes. Reads only.", "input": { "properties": { "seat": { @@ -250,7 +250,7 @@ }, { "name": "revision", - "description": "The catalog half of the revision view (ADR 0656 \u00a76): one catalog graph revision with its entries whole, the trunk heads, the seat definitions by digest, and each manifest's verdict and fingerprint; the trunk's newest when no revision and no candidate is named, or a candidate's for its group and heads. Answered by the holder from what it holds, in its own runtime (ADR 0663), in pages of entries. Reads only.", + "description": "The catalog half of the revision view (ADR 0656 §6): one catalog graph revision with its entries whole, the trunk heads, the seat definitions by digest, and each manifest's verdict and fingerprint; the trunk's newest when no revision and no candidate is named, or a candidate's for its group and heads. Answered by the holder from what it holds, in its own runtime (ADR 0663), in pages of entries. Reads only. A watched branch with no revision yet answers {none-yet: true} alone (design 47 G section 530).", "input": { "type": "object", "properties": { @@ -292,12 +292,12 @@ "next": { "type": "string", "description": "the module to ask after for the next page; absent on the last" + }, + "none-yet": { + "type": "boolean", + "description": "true for a watched branch with no revision yet: no revision, no entries and no error follow, and a reader holds the branch's view ready and empty (design 47 G section 530)" } - }, - "required": [ - "revision", - "entries" - ] + } }, "class": "read" }, @@ -554,17 +554,23 @@ "seat:node-build-node", "seat:node-build-image" ], - "description": "A step's input by its digest, as the indexer keeps it for a validation step it asked a build seat to run (design 47 G section 354).", + "description": "A step's input by its digest, as the indexer keeps it for a validation step it asked a build seat to run (design 47 G section 354): its fields, then the bytes in chunks of at most 512 KiB on the chunk subject of the call, each acknowledged by the caller on the ack subject within 30 seconds; the input is of any size.", "input": { "type": "object", "properties": { "digest": { "type": "string", "description": "the input's sha256" + }, + "call": { + "type": "string", + "description": "128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements", + "minLength": 1 } }, "required": [ - "digest" + "digest", + "call" ] }, "answer": { @@ -574,16 +580,18 @@ "type": "string", "description": "the input's sha256" }, - "input": { - "type": "string", - "description": "the input's exact bytes, base64" + "chunks": { + "type": "integer", + "minimum": 0, + "description": "how many chunks are sent; 0 when there is no input of that digest" } }, "required": [ "digest", - "input" + "chunks" ] - } + }, + "chunks": true }, { "name": "entry", @@ -614,7 +622,7 @@ } } ], - "description": "The catalog of the mesh: every module the trunk holds with its manifest, fingerprint and verdict, the packages published, and the seat list (ADR 0636 \u00a71, \u00a74).", + "description": "The catalog of the mesh: every module the trunk holds with its manifest, fingerprint and verdict, the packages published, and the seat list (ADR 0636 §1, §4).", "family": { "name": "mesh-catalog.events", "max-age-days": 7 @@ -678,6 +686,13 @@ "type": "integer", "description": "the indexer's revision of the verdict it says; a reader asks its first view again on a gap in it; zero from an indexer that does not say it" }, + "problems": { + "type": "array", + "items": { + "type": "string" + }, + "description": "each problem the indexer's re-judgement finds in a manifest it keeps accepted because its digest is unchanged: the manifest validator's strings, the same items its verdict writes as warnings, in the same order; an empty list when none (design 47 G section 668, 692)" + }, "bus-state": { "type": "array", "items": { @@ -717,7 +732,8 @@ "module", "digest", "judge", - "bus-state" + "bus-state", + "problems" ] } }, @@ -1785,7 +1801,7 @@ "name": "inputs", "template": "state", "history": 1, - "description": "a step's input, named by its write time and its sha256 (design 47 G section 354)", + "description": "a step's input, named by its write time and its sha256; a value over one chunk is held in the state's objects store (design 47 G sections 354, 712)", "key-forms": [ { "key": "