ADR 0627 §2: the sending of a file in parts is seat-neutral runtime; publish leaves mesh-sdk
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
This commit is contained in:
@@ -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",
|
||||
}
|
||||
|
||||
|
||||
@@ -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:<hex>), 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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:<hex>), 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:<seat>.<verb>`, 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
|
||||
}
|
||||
Reference in New Issue
Block a user