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

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",
 }
 

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

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 {
+	type receiver struct {
-		return func(key string, body any) (json.RawMessage, error) {
+		open  Open
-			*asked = append(*asked, key)
+		chunk Chunk
-			raw, _ := json.Marshal(body)
+	}
-			switch key {
+	holder := func(lacks []string, got *[]chunk, asked *[]string) receiver {
-			case "seat:oci-artifact-registry.publish":
+		return receiver{
-				return json.Marshal(registry.PublishAnswer{Transfer: "t1", Lacks: lacks, Held: len(lacks) == 0})
+			open: func(_ context.Context, ps []Part) (Opened, error) {
-			case "seat:oci-artifact-registry.publish-part":
+				*asked = append(*asked, "open")
-				var in registry.PublishPartInput
+				return Opened{Lacks: lacks, Held: len(lacks) == 0}, nil
-				_ = json.Unmarshal(raw, &in)
+			},
-				data, _ := base64.StdEncoding.DecodeString(in.Data)
+			chunk: func(_ context.Context, part string, seq int, last bool, data []byte) error {
-				if len(raw) > 1<<20 {
+				if len(data) > MaxChunk {
-					t.Errorf("a chunk's message is %d bytes, over the bus's largest", len(raw))
+					t.Errorf("a chunk is %d bytes, over MaxChunk", len(data))
 				}
-				*got = append(*got, chunk{in.Part, in.Sequence, in.Last, data})
+				*got = append(*got, chunk{part, int64(seq), last, append([]byte(nil), data...)})
-				return json.Marshal(registry.PublishPartAnswer{Transfer: in.Transfer, Part: in.Part})
+				return nil
-			}
+			},
-			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])}
 
@@ -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,
+	r := holder([]string{liar.Digest}, &got, &asked)
-		[]Part{liar}); err == nil || len(got) != 0 || !strings.Contains(err.Error(), "nothing of it is sent") {
+	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),
+	r = holder([]string{"sha256:" + strings.Repeat("f", 64)}, &got, &asked)
-		"oci-artifact-registry", in, parts); err == nil || !strings.Contains(err.Error(), "does not name") {
+	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)
 	}
 }

go/publish/publish.go

@@ -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
-}