From 0bb3a2c3927fa259965e881311eb9cb7d2395258 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:27:15 +0100 Subject: [PATCH] SDK-1: the runtime sends a file to a registry seat in parts and goes on from what the holder holds (ADR 0611, design 51) publish.Publish opens the transfer naming every part's digest and sends only the parts the holder answers it lacks, in chunks of at most 760 KiB (a message under the bus's 1 MiB once base64), in sequence with the last marked; a part whose bytes are not its digest is refused before any of it travels. The seat's messages are mesh-contracts'; the call is the caller's own. Its test takes the place of the vault's get test, merged into the vault's table. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/catalogue/vault_seat_test.go | 34 +------ go/packages_list_test.go | 2 +- go/publish/publish.go | 158 ++++++++++++++++++++++++++++++++ go/publish/publish_test.go | 122 ++++++++++++++++++++++++ 4 files changed, 284 insertions(+), 32 deletions(-) create mode 100644 go/publish/publish.go create mode 100644 go/publish/publish_test.go diff --git a/go/catalogue/vault_seat_test.go b/go/catalogue/vault_seat_test.go index 2d20e80..5b61d5e 100644 --- a/go/catalogue/vault_seat_test.go +++ b/go/catalogue/vault_seat_test.go @@ -27,37 +27,6 @@ func TestTheVaultsSeatDeliversSecret(t *testing.T) { } } -// The vault answers `get` on request and reply (novox/hq ADR 0349 §11, issue 674): the seat promises the -// verb, so a holder serving it is accepted — and, optional while the vault that runs is the TypeScript one, -// so is a holder that does not yet. -func TestTheVaultsSeatPromisesGet(t *testing.T) { - seat, _ := SeatNamed("mesh-vault") - var get *Verb - for i := range seat.Serves { - if seat.Serves[i].Name == "get" { - get = &seat.Serves[i] - } - } - if get == nil { - t.Fatal("the vault's seat does not promise get") - } - for _, arg := range []string{"name", "public_key"} { - if props, _ := get.Input["properties"].(map[string]any); props[arg] == nil { - t.Errorf("get takes no %s", arg) - } - } - serving := Manifest{Module: "mesh-vault", Provides: []Offer{{Name: "secret", Scope: ScopeMesh}}, - Claims: []Claim{{Name: "mesh-vault", Scope: ScopeMesh, Serves: []string{"get"}}}} - if err := CanHold(serving, seat); err != nil { - t.Fatalf("the vault serving get was refused: %v", err) - } - notYet := Manifest{Module: "mesh-vault", Provides: []Offer{{Name: "secret", Scope: ScopeMesh}}, - Claims: []Claim{{Name: "mesh-vault", Scope: ScopeMesh}}} - if err := CanHold(notYet, seat); err != nil { - t.Fatalf("the vault that does not serve get yet was refused: %v", err) - } -} - // The vault takes values in and rotates them (novox/hq ADR 0349 clauses 8 to 10, issue 675): the seat promises // secret-ask, issued and rotate, each with the arguments of its steps; a holder serving them is accepted, and, // optional while the vault that runs does not, so is one that does not yet. And the move of the controller's secrets @@ -66,6 +35,9 @@ func TestTheVaultsSeatPromisesGet(t *testing.T) { func TestTheVaultsSeatPromisesItsIntakeVerbs(t *testing.T) { seat, _ := SeatNamed("mesh-vault") want := map[string][]string{ + // get, on request and reply (ADR 0349 §11, issue 674); a vault not serving it yet still holds the seat + // (TestTheVaultsSeatDeliversSecret). + "get": {"name", "public_key"}, "secret-ask": {"node", "module", "name", "desk", "asked_by", "outcome", "call", "sealed"}, "issued": {"provision", "provider", "module", "call", "sealed", "holders"}, "rotate": {"node", "module", "name", "provider", "provision", "by"}, diff --git a/go/packages_list_test.go b/go/packages_list_test.go index cbef140..d0f3b04 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", + "objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions", "publish", "sdkversion", "seal", "seatdef", "testbus", "views", } diff --git a/go/publish/publish.go b/go/publish/publish.go new file mode 100644 index 0000000..30eead2 --- /dev/null +++ b/go/publish/publish.go @@ -0,0 +1,158 @@ +// 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' (parts/registry); 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" + + "git.novox.be/novox/mesh-contracts/go/parts/registry" +) + +// 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 +} diff --git a/go/publish/publish_test.go b/go/publish/publish_test.go new file mode 100644 index 0000000..08bf439 --- /dev/null +++ b/go/publish/publish_test.go @@ -0,0 +1,122 @@ +package publish + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "io" + "strings" + "testing" + + "git.novox.be/novox/mesh-contracts/go/parts/registry" +) + +// A file travels to a registry seat in parts (ADR 0611, 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. +func TestAFileTravelsInTheParts(t *testing.T) { + big := bytes.Repeat([]byte("x"), MaxChunk+10) + small, empty := []byte("config"), []byte{} + part := func(b []byte, opened *int) Part { + sum := sha256.Sum256(b) + return Part{Digest: "sha256:" + hex.EncodeToString(sum[:]), Size: int64(len(b)), + Open: func() (io.ReadCloser, error) { *opened++; return io.NopCloser(bytes.NewReader(b)), nil }} + } + type chunk struct { + part string + seq int64 + 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)) + } + *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 + } + } + 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])} + + for _, c := range []struct { + name string + lacks []int + chunks int + err string + }{ + {"a fresh transfer sends every part, the big one in two chunks and the empty one as one", []int{0, 1, 2}, 4, ""}, + {"a cut transfer opened again sends only what the holder lacks", []int{2}, 1, ""}, + {"a digest the holder holds sends nothing", nil, 0, ""}, + } { + opens = [3]int{} + var lacks []string + for _, i := range c.lacks { + lacks = append(lacks, parts[i].Digest) + } + var got []chunk + var asked []string + sent, err := Publish(context.Background(), holder(lacks, &got, &asked), "oci-artifact-registry", in, 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 + } + for i := range parts { + want := 0 + for _, l := range c.lacks { + if l == i { + want = 2 // read once to check its digest, once to send it + } + } + if opens[i] != want { + t.Errorf("%s: part %d read %d times, want %d", c.name, i, opens[i], want) + } + } + whole := map[string][]byte{} + for i, ch := range got { + whole[ch.part] = append(whole[ch.part], ch.data...) + lastOfPart := i == len(got)-1 || got[i+1].part != ch.part + if ch.last != lastOfPart || (i > 0 && got[i-1].part == ch.part && got[i-1].seq+1 != ch.seq) { + t.Errorf("%s: chunk %d of %s is %d, last %v", c.name, i, ch.part[:12], ch.seq, ch.last) + } + } + for _, l := range c.lacks { + if p := parts[l]; int64(len(whole[p.Digest])) != p.Size { + t.Errorf("%s: part %d arrived as %d bytes of %d", c.name, l, len(whole[p.Digest]), p.Size) + } + } + } + + // A part whose bytes are not its digest travels not at all; a part the holder names that the transfer does not is + // refused, naming it. + liar := part(small, &opens[1]) + 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") { + 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") { + t.Errorf("a part the transfer does not name: %v", err) + } +}