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) + } +}