From 32699e35b821f4560b7ce2267d4f4aea14301f49 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:15:50 +0100 Subject: [PATCH] =?UTF-8?q?CAT-D2:=20distribution's=20incoming=20binds=20m?= =?UTF-8?q?esh-sdk's=20seat-neutral=20receiving=20side=20(ADR=200627=20?= =?UTF-8?q?=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/distribution-tools/main.go | 4 +- .../internal/holder/holder_test.go | 6 +- modules/distribution/internal/holder/oci.go | 2 +- .../distribution/internal/holder/transfer.go | 317 +++++------------- 4 files changed, 88 insertions(+), 241 deletions(-) diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 9fa9423af..529b70c3c 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -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) diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index e910e671f..70cf7c685 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -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 diff --git a/modules/distribution/internal/holder/oci.go b/modules/distribution/internal/holder/oci.go index 23c1a5852..3c3e30b36 100644 --- a/modules/distribution/internal/holder/oci.go +++ b/modules/distribution/internal/holder/oci.go @@ -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 { diff --git a/modules/distribution/internal/holder/transfer.go b/modules/distribution/internal/holder/transfer.go index 97bd0c67a..a9c331f8c 100644 --- a/modules/distribution/internal/holder/transfer.go +++ b/modules/distribution/internal/holder/transfer.go @@ -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 `.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[:]) +}