From 1a149d44bef003383b504b66fda877731f6e19f3 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:09:19 +0100 Subject: [PATCH] =?UTF-8?q?ADR=200627=20=C2=A72:=20the=20sending=20of=20a?= =?UTF-8?q?=20file=20in=20parts=20is=20seat-neutral=20runtime;=20publish?= =?UTF-8?q?=20leaves=20mesh-sdk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit parts takes the open and the chunk calls from its caller and names no seat, verb or seat type; each registry seat's client in mesh-contracts binds it to its own seat (ADR 0627 §1). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/packages_list_test.go | 2 +- go/parts/parts.go | 145 ++++++++++++++++ .../publish_test.go => parts/parts_test.go} | 54 +++--- go/publish/publish.go | 159 ------------------ 4 files changed, 170 insertions(+), 190 deletions(-) create mode 100644 go/parts/parts.go rename go/{publish/publish_test.go => parts/parts_test.go} (61%) delete mode 100644 go/publish/publish.go diff --git a/go/packages_list_test.go b/go/packages_list_test.go index d0f3b04..a012bb4 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -25,7 +25,7 @@ import ( var meshSDKPackages = []string{ ".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership", - "objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions", "publish", + "objectlink", "outward", "overlay", "parts", "pernode", "process", "provider", "provider/harness", "provisions", "sdkversion", "seal", "seatdef", "testbus", "views", } diff --git a/go/parts/parts.go b/go/parts/parts.go new file mode 100644 index 0000000..e16c463 --- /dev/null +++ b/go/parts/parts.go @@ -0,0 +1,145 @@ +// Package parts is the runtime's sender of a file in parts (novox/hq ADR 0611, ADR 0627 §2, design 51 *Publishing*): +// the caller opens a transfer naming every part's digest and size, the receiver answers the parts it lacks, and only +// those travel, in chunks a bus message carries, in sequence with the last marked. A cut transfer goes on from what +// the receiver answers when the same transfer is opened again. It names no seat, no verb and no seat's type: the +// caller's Open and Chunk are its own calls (a registry seat's client binds them to that seat's verbs and types), so +// this package holds no transport and no contract. +package parts + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" +) + +// MaxChunk is the most bytes of a part one chunk carries: base64 makes it about 1 000 KiB, under 1 MiB, the server's +// largest message, with the message's own fields (design 51: each part travels in chunks of at most 1 MiB). +const MaxChunk = 760 << 10 + +// Part is one part of a transfer: its digest (sha256:), its size and how to read its bytes. +type Part struct { + Digest string + Size int64 + Open func() (io.ReadCloser, error) +} + +// Opened is the receiver's answer to opening a transfer: whether it holds the whole already, and the digests of the +// parts it lacks. +type Opened struct { + Held bool + Lacks []string +} + +// Open opens (or opens again) the transfer on the receiver, naming every part: the caller's own call. +type Open func(ctx context.Context, parts []Part) (Opened, error) + +// Chunk sends one chunk of a part: the part's digest, its sequence from 0, whether it is the part's last, its bytes. +type Chunk func(ctx context.Context, part string, sequence int, last bool, data []byte) error + +// Sent is what a Send did: whether the receiver held the whole, the parts it lacked and were sent, the chunks they +// took. +type Sent struct { + Held bool + Parts []string + Chunks int +} + +// Send opens the transfer and sends the parts the receiver answers it lacks, each in chunks of at most MaxChunk, in +// sequence, the last marked. A part whose bytes do not hash to its digest is refused before anything of it is sent; a +// part the receiver does not name is never read; a part the receiver names that the transfer does not is refused. +func Send(ctx context.Context, open Open, chunk Chunk, parts []Part) (Sent, error) { + byDigest := map[string]Part{} + for _, p := range parts { + byDigest[p.Digest] = p + } + opened, err := open(ctx, parts) + if err != nil { + return Sent{}, err + } + sent := Sent{Held: opened.Held} + for _, d := range opened.Lacks { + p, ok := byDigest[d] + if !ok { + return sent, fmt.Errorf("the receiver lacks the part %s, which the transfer does not name", d) + } + if err := ctx.Err(); err != nil { + return sent, err + } + if err := check(p); err != nil { + return sent, err + } + n, err := sendPart(ctx, chunk, p) + sent.Chunks += n + if err != nil { + return sent, err + } + sent.Parts = append(sent.Parts, d) + } + return sent, nil +} + +// check reads the part once and refuses it when its bytes are not its digest or its size. +func check(p Part) error { + r, err := p.Open() + if err != nil { + return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) + } + defer r.Close() + h := sha256.New() + n, err := io.Copy(h, r) + if err != nil { + return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) + } + if got := "sha256:" + hex.EncodeToString(h.Sum(nil)); got != p.Digest || n != p.Size { + return fmt.Errorf("the part %s holds %d bytes of digest %s, not its own %d: nothing of it is sent", p.Digest, n, got, p.Size) + } + return nil +} + +func sendPart(ctx context.Context, chunk Chunk, p Part) (int, error) { + r, err := p.Open() + if err != nil { + return 0, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) + } + defer r.Close() + buf := make([]byte, MaxChunk) + // A chunk is sent once the next read says whether it is the last, so the last is always marked: an empty part is + // one empty chunk, marked last. + var prev []byte + sent := 0 + send := func(data []byte, last bool) error { + if err := chunk(ctx, p.Digest, sent, last, data); err != nil { + return fmt.Errorf("chunk %d of the part %s: %w", sent, p.Digest, err) + } + sent++ + return nil + } + for { + if err := ctx.Err(); err != nil { + return sent, err + } + n, err := io.ReadFull(r, buf) + end := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) + if err != nil && !end { + return sent, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) + } + if end && n == 0 { + if prev == nil { + prev = []byte{} + } + return sent, send(prev, true) + } + if prev != nil { + if err := send(prev, false); err != nil { + return sent, err + } + } + prev = append([]byte(nil), buf[:n]...) + if end { + return sent, send(prev, true) + } + } +} diff --git a/go/publish/publish_test.go b/go/parts/parts_test.go similarity index 61% rename from go/publish/publish_test.go rename to go/parts/parts_test.go index 1e33c51..fa2e8da 100644 --- a/go/publish/publish_test.go +++ b/go/parts/parts_test.go @@ -1,20 +1,16 @@ -package publish +package parts import ( "bytes" "context" "crypto/sha256" - "encoding/base64" "encoding/hex" - "encoding/json" "io" "strings" "testing" - - registry "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" ) -// A file travels to a registry seat in parts (ADR 0611, design 51 *Publishing*; SDK-1): only the parts the holder says +// A file travels in parts (ADR 0611, ADR 0627 §2, design 51 *Publishing*; SDK-1): only the parts the holder says // it lacks are read and sent, each in chunks of at most MaxChunk in sequence with the last marked, so a cut transfer // opened again goes on from what the holder holds; a part whose bytes are not its digest is refused before any of it // travels. @@ -32,28 +28,25 @@ func TestAFileTravelsInTheParts(t *testing.T) { last bool data []byte } - holder := func(lacks []string, got *[]chunk, asked *[]string) Call { - return func(key string, body any) (json.RawMessage, error) { - *asked = append(*asked, key) - raw, _ := json.Marshal(body) - switch key { - case "seat:oci-artifact-registry.publish": - return json.Marshal(registry.PublishAnswer{Transfer: "t1", Lacks: lacks, Held: len(lacks) == 0}) - case "seat:oci-artifact-registry.publish-part": - var in registry.PublishPartInput - _ = json.Unmarshal(raw, &in) - data, _ := base64.StdEncoding.DecodeString(in.Data) - if len(raw) > 1<<20 { - t.Errorf("a chunk's message is %d bytes, over the bus's largest", len(raw)) + type receiver struct { + open Open + chunk Chunk + } + holder := func(lacks []string, got *[]chunk, asked *[]string) receiver { + return receiver{ + open: func(_ context.Context, ps []Part) (Opened, error) { + *asked = append(*asked, "open") + return Opened{Lacks: lacks, Held: len(lacks) == 0}, nil + }, + chunk: func(_ context.Context, part string, seq int, last bool, data []byte) error { + if len(data) > MaxChunk { + t.Errorf("a chunk is %d bytes, over MaxChunk", len(data)) } - *got = append(*got, chunk{in.Part, in.Sequence, in.Last, data}) - return json.Marshal(registry.PublishPartAnswer{Transfer: in.Transfer, Part: in.Part}) - } - t.Fatalf("asked %s", key) - return nil, nil + *got = append(*got, chunk{part, int64(seq), last, append([]byte(nil), data...)}) + return nil + }, } } - in := registry.PublishInput{Transfer: "t1", Name: "files", Kind: "archive", Digest: "sha256:" + strings.Repeat("a", 64)} var opens [3]int parts := []Part{part(big, &opens[0]), part(small, &opens[1]), part(empty, &opens[2])} @@ -74,7 +67,8 @@ func TestAFileTravelsInTheParts(t *testing.T) { } var got []chunk var asked []string - sent, err := Publish(context.Background(), holder(lacks, &got, &asked), "oci-artifact-registry", in, parts) + r := holder(lacks, &got, &asked) + sent, err := Send(context.Background(), r.open, r.chunk, parts) if err != nil || sent.Chunks != c.chunks || len(got) != c.chunks || len(sent.Parts) != len(c.lacks) { t.Errorf("%s: sent %+v, %d chunks heard, %v", c.name, sent, len(got), err) continue @@ -111,12 +105,12 @@ func TestAFileTravelsInTheParts(t *testing.T) { liar.Digest = parts[0].Digest var got []chunk var asked []string - if _, err := Publish(context.Background(), holder([]string{liar.Digest}, &got, &asked), "oci-artifact-registry", in, - []Part{liar}); err == nil || len(got) != 0 || !strings.Contains(err.Error(), "nothing of it is sent") { + r := holder([]string{liar.Digest}, &got, &asked) + if _, err := Send(context.Background(), r.open, r.chunk, []Part{liar}); err == nil || len(got) != 0 || !strings.Contains(err.Error(), "nothing of it is sent") { t.Errorf("a part not its digest: %v, %d chunks sent", err, len(got)) } - if _, err := Publish(context.Background(), holder([]string{"sha256:" + strings.Repeat("f", 64)}, &got, &asked), - "oci-artifact-registry", in, parts); err == nil || !strings.Contains(err.Error(), "does not name") { + r = holder([]string{"sha256:" + strings.Repeat("f", 64)}, &got, &asked) + if _, err := Send(context.Background(), r.open, r.chunk, parts); err == nil || !strings.Contains(err.Error(), "does not name") { t.Errorf("a part the transfer does not name: %v", err) } } diff --git a/go/publish/publish.go b/go/publish/publish.go deleted file mode 100644 index 3503fb1..0000000 --- a/go/publish/publish.go +++ /dev/null @@ -1,159 +0,0 @@ -// Package publish is the runtime's helper for sending a file to a registry seat in parts (novox/hq ADR 0611, design -// 51 *Publishing*): `publish` opens a transfer naming every part's digest, the holder answers the parts it lacks, and -// only those travel, by `publish-part`, in chunks a bus message carries. A cut transfer goes on from what the holder -// answers when the same transfer is opened again. The seat's messages are mesh-contracts' -// (the registry part's, read through containerimageregistry, the four registry seats' shapes being one); the call is -// the caller's own (a bundle's stdio.Ask, a process's call of a seat verb), so this package holds no transport. -package publish - -import ( - "context" - "crypto/sha256" - "encoding/base64" - "encoding/hex" - "encoding/json" - "errors" - "fmt" - "io" - - registry "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" -) - -// MaxChunk is the most bytes of a part one publish-part carries: base64 makes it about 1 000 KiB, under 1 MiB, the -// server's largest message, with the message's own fields (design 51: each part travels in chunks of at most 1 MiB). -const MaxChunk = 760 << 10 - -// Part is one part of a transfer: its digest (sha256:), its size and how to read its bytes. -type Part struct { - Digest string - Size int64 - Open func() (io.ReadCloser, error) -} - -// Call asks one verb of the seat, `seat:.`, with a body, and answers its JSON. -type Call func(key string, body any) (json.RawMessage, error) - -// Sent is what a Publish did: the parts the holder lacked and were sent, and the chunks they took. -type Sent struct { - Transfer string - Held bool - Parts []string - Chunks int -} - -// Publish opens (or opens again) the transfer in on the seat and sends the parts the holder answers it lacks, each in -// chunks of at most MaxChunk, in sequence, the last marked. A part whose bytes do not hash to its digest is refused -// before anything of it is sent; a part the holder does not name is never read. Its outcome, published or refused, is -// the holder's event, which the caller waits for. -func Publish(ctx context.Context, call Call, seat string, in registry.PublishInput, parts []Part) (Sent, error) { - in.Parts = in.Parts[:0] - byDigest := map[string]Part{} - for _, p := range parts { - in.Parts = append(in.Parts, registry.Part{Digest: p.Digest, Size: p.Size}) - byDigest[p.Digest] = p - } - raw, err := call("seat:"+seat+"."+registry.VerbPublish, in) - if err != nil { - return Sent{}, fmt.Errorf("%s's publish of %s: %w", seat, in.Name, err) - } - var opened registry.PublishAnswer - if err := json.Unmarshal(raw, &opened); err != nil { - return Sent{}, fmt.Errorf("%s's answer to publish: %w", seat, err) - } - sent := Sent{Transfer: in.Transfer, Held: opened.Held} - for _, d := range opened.Lacks { - p, ok := byDigest[d] - if !ok { - return sent, fmt.Errorf("%s lacks the part %s, which the transfer %s does not name", seat, d, in.Transfer) - } - if err := ctx.Err(); err != nil { - return sent, err - } - if err := check(p); err != nil { - return sent, err - } - n, err := sendPart(ctx, call, seat, in.Transfer, p) - sent.Chunks += n - if err != nil { - return sent, err - } - sent.Parts = append(sent.Parts, d) - } - return sent, nil -} - -// check reads the part once and refuses it when its bytes are not its digest or its size. -func check(p Part) error { - r, err := p.Open() - if err != nil { - return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) - } - defer r.Close() - h := sha256.New() - n, err := io.Copy(h, r) - if err != nil { - return fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) - } - if got := "sha256:" + hex.EncodeToString(h.Sum(nil)); got != p.Digest || n != p.Size { - return fmt.Errorf("the part %s holds %d bytes of digest %s, not its own %d: nothing of it is sent", p.Digest, n, got, p.Size) - } - return nil -} - -func sendPart(ctx context.Context, call Call, seat, transfer string, p Part) (int, error) { - r, err := p.Open() - if err != nil { - return 0, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) - } - defer r.Close() - buf := make([]byte, MaxChunk) - // A chunk is sent once the next read says whether it is the last, so the last is always marked: an empty part is - // one empty chunk, marked last. - var prev []byte - sent := 0 - send := func(data []byte, last bool) error { - if err := sendChunk(call, seat, transfer, p.Digest, sent, last, data); err != nil { - return err - } - sent++ - return nil - } - for { - if err := ctx.Err(); err != nil { - return sent, err - } - n, err := io.ReadFull(r, buf) - end := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) - if err != nil && !end { - return sent, fmt.Errorf("the part %s cannot be read: %w", p.Digest, err) - } - if end && n == 0 { - if prev == nil { - prev = []byte{} - } - return sent, send(prev, true) - } - if prev != nil { - if err := send(prev, false); err != nil { - return sent, err - } - } - prev = append([]byte(nil), buf[:n]...) - if end { - return sent, send(prev, true) - } - } -} - -func sendChunk(call Call, seat, transfer, part string, sequence int, last bool, data []byte) error { - raw, err := call("seat:"+seat+"."+registry.VerbPublishPart, registry.PublishPartInput{Transfer: transfer, Part: part, - Sequence: int64(sequence), Last: last, Data: base64.StdEncoding.EncodeToString(data)}) - if err != nil { - return fmt.Errorf("%s's publish-part %d of %s: %w", seat, sequence, part, err) - } - var a registry.PublishPartAnswer - if err := json.Unmarshal(raw, &a); err != nil { - return fmt.Errorf("%s's answer to publish-part: %w", seat, err) - } - return nil -}