novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

CAT-D2: distribution's incoming binds mesh-sdk's seat-neutral receiving side (ADR 0627 §2)

Browse Source 
The holder's own transfer logic goes: parts.Receiver takes the chunks into incoming, checks the hashes, drops an idle
transfer and tells the fill; distribution gives it the builder_versions check, its store and its seats' events, so
the forge's two package seats bind the same receiver instead of a copy.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 13:15:50 +01:00
parent 85fd46b4cc
commit 32699e35b8
4 changed files with 88 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/distribution/cmd/distribution-tools/main.go
+2 -2
View File
@@ -309,7 +309,7 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
}
t, err := v.Drop(id)
logf("transfer %s dropped by hand by %s: %s", id, stdio.CallerOf(args), str(args, "why"))
return map[string]any{"dropped": t.Transfer}, err
return map[string]any{"dropped": t.ID}, err
}
all, err := v.Transfers.List()
used, _ := v.Objects.Used()
@@ -369,7 +369,7 @@ func main() {
base, _ := url.Parse("http://127.0.0.1:5001")
store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH"), Upstream: &holder.Upstream{}}
h := &holder.Holder{Callers: callers, Objects: objects{}, Builder: builderVersions{}, Store: store, Say: say, Finding: finding,
Transfers: kept[holder.Transfer]{state: holder.TransfersState, key: func(t holder.Transfer) string { return t.Transfer }}}
Transfers: kept[holder.Transfer]{state: holder.TransfersState, key: func(t holder.Transfer) string { return t.ID }}}
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}}
windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
closeBootstrap = bootstrapCloser(windows)
modules/distribution/internal/holder/holder_test.go
+3 -3
View File
@@ -37,7 +37,7 @@ func (m memObjects) Used() (uint64, error) {
type memTransfers map[string]Transfer
func (m memTransfers) Get(id string) (Transfer, bool, error) { t, ok := m[id]; return t, ok, nil }
func (m memTransfers) Put(t Transfer) error { m[t.Transfer] = t; return nil }
func (m memTransfers) Put(t Transfer) error { m[t.ID] = t; return nil }
func (m memTransfers) Delete(id string) error { delete(m, id); return nil }
func (m memTransfers) List() ([]Transfer, error) {
var out []Transfer
@@ -62,7 +62,7 @@ type fakeStore struct {
}
func (s *fakeStore) Has(_, d string) (bool, error) { return s.has[d], nil }
func (s *fakeStore) Put(t Transfer, parts map[string][]byte) (string, error) {
func (s *fakeStore) Put(t PublishRequest, parts map[string][]byte) (string, error) {
s.put = append(s.put, parts)
return "registry.mesh/" + t.Name + "@" + t.Digest, nil
}
@@ -207,7 +207,7 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
_, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t", Digest: part.Digest, Data: make([]byte, MaxPart+1)})
return err
}, "over the"},
{"a new transfer at the cap", func(b *bed) error { b.h.Cap = 100; _, err := b.h.Publish(ArtifactSeat, builder, req("t")); return err }, "at its cap"},
{"a new transfer at the cap", func(b *bed) error { b.h.Cap = 100; _, err := b.h.Publish(ArtifactSeat, builder, req("t")); return err }, "at their cap"},
{"a version the builder did not number", func(b *bed) error {
b.h.Builder = fakeBuilder{ok: false}
b.store.has[part.Digest] = true // held already: the publish ends at once
modules/distribution/internal/holder/oci.go
+1 -1
View File
@@ -110,7 +110,7 @@ func (o *OCI) putManifest(name, ref, mediaType string, data []byte) (string, err
// Put stores a transfer: an image's layers and config as blobs and its manifest (the part whose media type is a
// manifest's) under its version; a bundle's or an archive's one blob under an artifact manifest of an empty config and
// one layer (the shape the artifact sweep recognises), under its version, or by its digest alone when unnumbered.
func (o *OCI) Put(t Transfer, parts map[string][]byte) (string, error) {
func (o *OCI) Put(t PublishRequest, parts map[string][]byte) (string, error) {
var manifest *Part
digests := make([]string, 0, len(parts))
for d := range parts {
modules/distribution/internal/holder/transfer.go
+82 -235
View File
@@ -1,15 +1,15 @@
package holder
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/parts"
)
// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, the store 20 GiB, a transfer
@@ -72,28 +72,15 @@ type PartRequest struct {
Data []byte `json:"data"`
}
// Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming.
type Transfer struct {
// Opening is what distribution keeps of a publish beside the receiving side's own fields: the request and its seat.
type Opening struct {
PublishRequest
Seat string `json:"seat"`
Opener string `json:"opener"`
// Held are the parts whose every chunk arrived and hashed right; Chunks the chunks heard per part.
Held []string `json:"held,omitempty"`
Chunks map[string]int `json:"chunks,omitempty"`
Opened time.Time `json:"opened"`
Last time.Time `json:"last"`
Seat string `json:"seat"`
}
func (t Transfer) missing() []string {
var out []string
for _, p := range t.Parts {
if !slices.Contains(t.Held, p.Digest) && !slices.Contains(out, p.Digest) {
out = append(out, p.Digest)
}
}
sort.Strings(out)
return out
}
// Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming. The receiving side is
// mesh-sdk's seat-neutral parts.Receiver (ADR 0627 §2); distribution binds it to the seats it holds.
type Transfer = parts.Transfer[Opening]
// Event is the body of `published`, `refused` and `removed` on `<seat>.events` (design 51 *One contract*): the name,
// version, kind, tag, digest, fingerprint, commit and holder; `refused` its reason, `removed` its reason too.
@@ -124,21 +111,10 @@ const (
)
// Objects is distribution's own object store incoming.
type Objects interface {
Put(name string, data []byte) error
Get(name string) ([]byte, error)
Delete(name string) error
// Used is the bytes the store holds.
Used() (uint64, error)
}
type Objects = parts.Objects
// TransferState is the state transfers.
type TransferState interface {
Get(id string) (Transfer, bool, error)
Put(t Transfer) error
Delete(id string) error
List() ([]Transfer, error)
}
type TransferState = parts.Transfers[Opening]
// Builder is the builder's tool builder_versions, asked before anything is stored (ADR 0605 §4).
type Builder interface {
@@ -149,8 +125,8 @@ type Builder interface {
type Store interface {
// Has says whether the store holds a blob of name.
Has(name, digest string) (bool, error)
// Put stores a transfer's parts under its name and version and answers the reference.
Put(t Transfer, parts map[string][]byte) (reference string, err error)
// Put stores a publish's parts under its name and version and answers the reference.
Put(r PublishRequest, parts map[string][]byte) (reference string, err error)
// Held is the reference of name at version when the store already holds digest under it: a rebuild of a held
// fingerprint answers it and stores nothing.
Held(name, version, digest string) (reference string, ok bool, err error)
@@ -190,6 +166,55 @@ func (h *Holder) cap() uint64 {
return IncomingCap
}
// receiver binds mesh-sdk's receiving side to distribution's incoming, its transfers state, the builder's check, its
// store and its seats' events.
func (h *Holder) receiver() *parts.Receiver[Opening] {
return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: h.cap(), MaxChunk: MaxPart,
Idle: TransferIdle, Full: IncomingFullPct, Now: h.now,
// Only the missing layers are sent: a part the store already holds under the name counts as held.
Holds: func(t Transfer, digest string) bool {
ok, err := h.Store.Has(t.Shape.Name, digest)
return err == nil && ok
},
// builder_versions is asked before anything is stored (ADR 0605 §4).
Check: func(t Transfer) error {
if t.Shape.Version == "" {
return nil
}
ok, why, err := h.Builder.Recorded(t.Shape.Name, t.Shape.Version, t.Shape.Digest)
if err != nil {
return errors.New("builder_versions did not answer, so nothing is stored: " + err.Error())
}
if !ok {
return errors.New("the builder did not number it: " + why)
}
return nil
},
Store: func(t Transfer, held map[string][]byte) error {
ref, err := h.Store.Put(t.Shape.PublishRequest, held)
if err != nil {
reason := "the store did not take it: " + err.Error()
return errors.Join(errors.New("refused: "+reason), h.refused(t, reason))
}
h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, "", false)
return h.Say(t.Shape.Seat, EventPublished, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, ref, ""))
},
Refuse: h.refused,
Filled: func(used, cap uint64, full bool) {
h.Finding("incoming-full", IncomingObjects, fmt.Sprintf("incoming holds %d of its %d bytes (%d %%); registry_incoming lists and drops transfers", used, cap, used*100/cap), full)
}}
}
// refused writes the finding and says refused on the transfer's seat.
func (h *Holder) refused(t Transfer, reason string) error {
h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, reason, true)
return h.Say(t.Shape.Seat, EventRefused, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, "", reason))
}
func answerOf(r parts.Received) PublishAnswer {
return PublishAnswer{Transfer: r.ID, Missing: r.Lacks, Ended: r.Ended}
}
// Publish is `publish` on seat for the proved caller.
func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer, error) {
if err := h.Callers.Allowed(VerbPublish, caller); err != nil {
@@ -198,15 +223,6 @@ func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer,
if err := checkPublish(seat, r); err != nil {
return PublishAnswer{}, err
}
if t, found, err := h.Transfers.Get(r.Transfer); err != nil {
return PublishAnswer{}, err
} else if found {
// A publish opening a transfer already open answers the parts held: a cut transfer goes on from there.
if t.Opener != caller {
return PublishAnswer{}, fmt.Errorf("publish: refused for %s: transfer %s was opened by another caller", caller, r.Transfer)
}
return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
}
// A rebuild of a held fingerprint answers the held version and stores nothing.
if r.Version != "" {
if ref, ok, err := h.Store.Held(r.Name, r.Version, r.Digest); err != nil {
@@ -215,34 +231,15 @@ func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer,
return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.Say(seat, EventPublished, h.eventOf(seat, r, ref, ""))
}
}
used, err := h.Objects.Used()
wants := make([]parts.Want, 0, len(r.Parts))
for _, p := range r.Parts {
wants = append(wants, parts.Want{Digest: p.Digest, Size: p.Size})
}
got, err := h.receiver().Open(r.Transfer, caller, wants, Opening{PublishRequest: r, Seat: seat})
if err != nil {
return PublishAnswer{}, fmt.Errorf("publish: incoming's fill cannot be read: %w", err)
return answerOf(got), fmt.Errorf("publish: %w", err)
}
var size uint64
for _, p := range r.Parts {
size += uint64(max(p.Size, 0))
}
if used+size > h.cap() {
reason := fmt.Sprintf("incoming is at its cap of %d bytes (%d held, %d asked)", h.cap(), used, size)
_ = h.Say(seat, EventRefused, h.eventOf(seat, r, "", reason))
return PublishAnswer{Transfer: r.Transfer, Ended: true}, errors.New("publish: refused: " + reason)
}
t := Transfer{PublishRequest: r, Seat: seat, Opener: caller, Opened: h.now(), Last: h.now(), Chunks: map[string]int{}}
// Only the missing layers are sent: a part the store already holds under the name counts as held.
for _, p := range r.Parts {
if ok, err := h.Store.Has(r.Name, p.Digest); err == nil && ok {
t.Held = append(t.Held, p.Digest)
}
}
if err := h.Transfers.Put(t); err != nil {
return PublishAnswer{}, fmt.Errorf("publish: transfer %s cannot be kept: %w", r.Transfer, err)
}
h.fill()
if len(t.missing()) == 0 {
return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.finish(t)
}
return PublishAnswer{Transfer: r.Transfer, Missing: t.missing()}, nil
return answerOf(got), nil
}
// PublishPart is `publish-part`: one chunk, from the caller that opened the transfer alone.
@@ -250,95 +247,16 @@ func (h *Holder) PublishPart(seat, caller string, r PartRequest) (PublishAnswer,
if err := h.Callers.Allowed(VerbPublishPart, caller); err != nil {
return PublishAnswer{}, err
}
t, found, err := h.Transfers.Get(r.Transfer)
switch {
case err != nil:
return PublishAnswer{}, err
case !found:
return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s is not open (dropped after %s with no part, or never opened by publish)", r.Transfer, TransferIdle)
case t.Opener != caller:
return PublishAnswer{}, fmt.Errorf("publish-part: refused for %s: transfer %s was opened by another caller", caller, r.Transfer)
case t.Seat != seat:
return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s is %s's, not %s's", r.Transfer, t.Seat, seat)
case len(r.Data) > MaxPart:
return PublishAnswer{}, fmt.Errorf("publish-part: a chunk of %d bytes is over the %d a part may carry", len(r.Data), MaxPart)
case !slices.ContainsFunc(t.Parts, func(p Part) bool { return p.Digest == r.Digest }):
return PublishAnswer{}, fmt.Errorf("publish-part: transfer %s names no part %s", r.Transfer, r.Digest)
case slices.Contains(t.Held, r.Digest):
return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
case r.Sequence != t.Chunks[r.Digest]:
return PublishAnswer{}, fmt.Errorf("publish-part: part %s of transfer %s waits for chunk %d, not %d", r.Digest, r.Transfer, t.Chunks[r.Digest], r.Sequence)
}
if err := h.Objects.Put(chunkName(t.Transfer, r.Digest, r.Sequence), r.Data); err != nil {
return PublishAnswer{}, fmt.Errorf("publish-part: incoming did not take the chunk: %w", err)
}
t.Chunks[r.Digest] = r.Sequence + 1
t.Last = h.now()
if r.Last {
whole, err := h.part(t, r.Digest)
if err != nil {
return PublishAnswer{}, err
got, err := h.receiver().Take(r.Transfer, caller, r.Digest, r.Sequence, r.Last, r.Data, func(t Transfer) error {
if t.Shape.Seat != seat {
return fmt.Errorf("transfer %s is %s's, not %s's", r.Transfer, t.Shape.Seat, seat)
}
if got := "sha256:" + sum(whole); got != r.Digest {
h.dropChunks(t, r.Digest)
delete(t.Chunks, r.Digest)
_ = h.Transfers.Put(t)
return PublishAnswer{}, fmt.Errorf("publish-part: refused: the bytes of part %s hash to %s; send the part again", r.Digest, got)
}
t.Held = append(t.Held, r.Digest)
}
if err := h.Transfers.Put(t); err != nil {
return PublishAnswer{}, err
}
h.fill()
if len(t.missing()) == 0 {
return PublishAnswer{Transfer: t.Transfer, Ended: true}, h.finish(t)
}
return PublishAnswer{Transfer: t.Transfer, Missing: t.missing()}, nil
}
// finish asks the builder and stores, and says published or refused; the transfer's chunks go either way.
func (h *Holder) finish(t Transfer) error {
defer func() {
for _, p := range t.Parts {
h.dropChunks(t, p.Digest)
}
_ = h.Transfers.Delete(t.Transfer)
h.fill()
}()
refuse := func(reason string) error {
h.Finding("publish-refused", t.Name+"@"+t.Version, reason, true)
if err := h.Say(t.Seat, EventRefused, h.eventOf(t.Seat, t.PublishRequest, "", reason)); err != nil {
return fmt.Errorf("%s; and refused could not be said: %w", reason, err)
}
return errors.New("publish: refused: " + reason)
}
if t.Version != "" {
ok, why, err := h.Builder.Recorded(t.Name, t.Version, t.Digest)
if err != nil {
return refuse("builder_versions did not answer, so nothing is stored: " + err.Error())
}
if !ok {
return refuse("the builder did not number it: " + why)
}
}
parts := map[string][]byte{}
for _, p := range t.Parts {
if _, had := t.Chunks[p.Digest]; !had {
continue // held by the store already
}
b, err := h.part(t, p.Digest)
if err != nil {
return refuse("part " + p.Digest + " cannot be read from incoming: " + err.Error())
}
parts[p.Digest] = b
}
ref, err := h.Store.Put(t, parts)
return nil
})
if err != nil {
return refuse("the store did not take it: " + err.Error())
return answerOf(got), fmt.Errorf("publish-part: %w", err)
}
h.Finding("publish-refused", t.Name+"@"+t.Version, "", false)
return h.Say(t.Seat, EventPublished, h.eventOf(t.Seat, t.PublishRequest, ref, ""))
return answerOf(got), nil
}
func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Event {
@@ -346,92 +264,21 @@ func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Even
Commit: r.Commit, Holder: Self, Transfer: r.Transfer, Reference: ref, Reason: reason}
}
func (h *Holder) part(t Transfer, digest string) ([]byte, error) {
var buf bytes.Buffer
for seq := 0; seq < t.Chunks[digest]; seq++ {
b, err := h.Objects.Get(chunkName(t.Transfer, digest, seq))
if err != nil {
return nil, err
}
buf.Write(b)
}
return buf.Bytes(), nil
}
func (h *Holder) dropChunks(t Transfer, digest string) {
for seq := 0; seq < t.Chunks[digest]; seq++ {
_ = h.Objects.Delete(chunkName(t.Transfer, digest, seq))
}
}
// fill writes incoming's finding at 80 % of the cap and clears it below.
func (h *Holder) fill() {
used, err := h.Objects.Used()
if err != nil {
return
}
full := used*100 >= h.cap()*IncomingFullPct
h.Finding("incoming-full", IncomingObjects, fmt.Sprintf("incoming holds %d of its %d bytes (%d %%); registry_incoming lists and drops transfers", used, h.cap(), used*100/h.cap()), full)
}
// Drop drops a transfer by hand (registry_incoming) or because it was idle for TransferIdle: its chunks and its key.
func (h *Holder) Drop(id string) (Transfer, error) {
t, found, err := h.Transfers.Get(id)
if err != nil {
return Transfer{}, err
}
if !found {
return Transfer{}, fmt.Errorf("transfer %s is not open", id)
}
for _, p := range t.Parts {
h.dropChunks(t, p.Digest)
}
if err := h.Transfers.Delete(id); err != nil {
return Transfer{}, err
}
h.fill()
return t, nil
}
func (h *Holder) Drop(id string) (Transfer, error) { return h.receiver().Drop(id) }
// DropIdle drops every transfer with no part for TransferIdle, answering them; it runs on each publish's open and each
// registry_incoming call, never on a timer (the object store's own lifetime of 24 hours drops the bytes too).
func (h *Holder) DropIdle() ([]string, error) {
all, err := h.Transfers.List()
if err != nil {
return nil, err
}
var out []string
for _, t := range all {
if h.now().Sub(t.Last) >= TransferIdle {
if _, err := h.Drop(t.Transfer); err == nil {
out = append(out, t.Transfer)
}
}
}
return out, nil
}
func chunkName(transfer, digest string, seq int) string {
return fmt.Sprintf("%s/%s/%d", transfer, strings.TrimPrefix(digest, "sha256:"), seq)
}
func sum(b []byte) string {
s := sha256.Sum256(b)
return hex.EncodeToString(s[:])
}
func (h *Holder) DropIdle() ([]string, error) { return h.receiver().DropIdle() }
func checkPublish(seat string, r PublishRequest) error {
switch {
case !slices.Contains(Seats, seat):
return fmt.Errorf("publish: distribution holds no seat %s", seat)
case strings.TrimSpace(r.Transfer) == "" || strings.ContainsAny(r.Transfer, "/. *>"):
return fmt.Errorf("publish: transfer %q is no transfer id", r.Transfer)
case strings.TrimSpace(r.Name) == "":
return errors.New("publish: no name")
case !strings.HasPrefix(r.Digest, "sha256:"):
return fmt.Errorf("publish: digest %q is no sha256 digest", r.Digest)
case len(r.Parts) == 0:
return errors.New("publish: no parts")
}
switch {
case seat == ImageSeat && r.Kind != KindImage && r.Kind != KindUpstream:
@@ -439,10 +286,10 @@ func checkPublish(seat string, r PublishRequest) error {
case seat == ArtifactSeat && r.Kind != KindBundle && r.Kind != KindArchive:
return fmt.Errorf("publish: %s holds bundles and archives, not a %s", seat, r.Kind)
}
for _, p := range r.Parts {
if !strings.HasPrefix(p.Digest, "sha256:") || len(p.Digest) != len("sha256:")+64 {
return fmt.Errorf("publish: part digest %q is no sha256 digest", p.Digest)
}
}
return nil
}
func sum(b []byte) string {
s := sha256.Sum256(b)
return hex.EncodeToString(s[:])
}