diff --git a/modules/distribution/go.mod b/modules/distribution/go.mod new file mode 100644 index 000000000..be2ed40e3 --- /dev/null +++ b/modules/distribution/go.mod @@ -0,0 +1,3 @@ +module distribution + +go 1.26.0 diff --git a/modules/distribution/go.sum b/modules/distribution/go.sum new file mode 100644 index 000000000..e69de29bb diff --git a/modules/distribution/internal/holder/callers.go b/modules/distribution/internal/holder/callers.go new file mode 100644 index 000000000..1078d6eb6 --- /dev/null +++ b/modules/distribution/internal/holder/callers.go @@ -0,0 +1,131 @@ +// Package holder is distribution's holding of the two image seats, `container-image-registry` and +// `oci-artifact-registry`, and of the old seat `mesh-artifact-store` until its window closes (novox/hq design 51 *The +// four registry seats*, ADR 0605, ADR 0611, ADR 0617 §2). A build module publishes into a seat through `publish` and +// `publish-part`: the parts land in distribution's own object store `incoming`, are hashed, checked against the +// builder's `builder_versions` and stored in the artifact store, which listens on its node's loopback; the store's port +// is the program's, reads open, writes from its own process alone (front.go). +package holder + +import ( + "encoding/json" + "fmt" + "slices" + "sort" + "strings" +) + +// The seats distribution holds, and the old one it holds until its window closes. +const ( + ImageSeat = "container-image-registry" + ArtifactSeat = "oci-artifact-registry" + OldSeat = "mesh-artifact-store" +) + +// The verbs of the seats (design 26 *A seat's definition*; design 51 *One contract*). +const ( + VerbPublish = "publish" + VerbPublishPart = "publish-part" + VerbVersions = "versions" + VerbRemove = "remove" + VerbMirror = "mirror" + VerbTags = "tags" + VerbHold = "hold" + VerbRelease = "release" +) + +// Seats are the seats distribution answers verbs on. +var Seats = []string{ImageSeat, ArtifactSeat} + +// VerbsOf are a seat's verbs: mirror and tags on the image seat alone; hold and release on both of distribution's. +func VerbsOf(seat string) []string { + v := []string{VerbPublish, VerbPublishPart, VerbVersions, VerbRemove, VerbHold, VerbRelease} + if seat == ImageSeat { + v = append(v, VerbMirror, VerbTags) + } + sort.Strings(v) + return v +} + +// Callers is the holder's trusted config value `callers` (the form ADR 0363 gives a restricted family): per verb, the +// modules that may call it. The holder checks the caller the tool runner proved and stamped on the request (design 25 +// § 12.4), never a name in the request's body. +type Callers map[string][]string + +// DefaultCallers is the list design 51 *Who may call which verb* names: publish and publish-part the build modules; +// mirror mesh-build-image; remove the builder (a tagged version) and the artifact sweep's holder (retention); hold the +// artifact sweep's holder alone; release that holder, and the operator's session only through registry_holds (Operator +// below). versions is read by anyone on the mesh, as the store's reads are. +func DefaultCallers() Callers { + builds := []string{"mesh-build-go", "mesh-build-image", "mesh-build-node"} + return Callers{ + VerbPublish: builds, + VerbPublishPart: builds, + VerbMirror: {"mesh-build-image"}, + VerbRemove: {"mesh-delivery-builder", SweepHolder}, + VerbHold: {SweepHolder}, + VerbRelease: {SweepHolder, Operator}, + VerbVersions: {Anyone}, + VerbTags: {Anyone}, + } +} + +// SweepHolder is the module holding the seat `mesh-artifact-sweep`, the artifact sweep. +const SweepHolder = "artifact-store-tools" + +// Anyone on a verb's list lets every proved caller call it. +const Anyone = "*" + +// Operator on a list stands for the operator's session, the caller the tool runner proves as `person.` (ADR 0386 +// §7): a release by hand through registry_holds carries it on, never distribution's own account. +const Operator = "operator" + +// Unproven is the caller the runtime names for a call it could not prove (mesh-sdk stdio.Unproven). +const Unproven = "unproven" + +// Self is distribution's own account: never on release's list, so registry_holds cannot release in its own name. +const Self = "distribution" + +// ParseCallers reads the config value; an empty value is DefaultCallers. A verb no seat has is refused, and Self on +// release's list is refused (design 51). +func ParseCallers(raw string) (Callers, error) { + if strings.TrimSpace(raw) == "" { + return DefaultCallers(), nil + } + var c Callers + if err := json.Unmarshal([]byte(raw), &c); err != nil { + return nil, fmt.Errorf("callers is not a JSON object of verb to modules: %w", err) + } + known := VerbsOf(ImageSeat) + for verb, mods := range c { + if !slices.Contains(known, verb) { + return nil, fmt.Errorf("callers names the verb %q, which no seat of distribution's has", verb) + } + if verb == VerbRelease && slices.Contains(mods, Self) { + return nil, fmt.Errorf("callers puts %s on release's list: registry_holds would release in its own name", Self) + } + } + return c, nil +} + +// Allowed answers nil when caller may call verb, else the refusal naming the verb and the caller. +func (c Callers) Allowed(verb, caller string) error { + if caller == "" || caller == Unproven { + return fmt.Errorf("%s: refused: the request carries no caller the tool runner proved", verb) + } + person := strings.HasPrefix(caller, "person.") + for _, m := range c[verb] { + switch { + case m == Anyone, m == Operator && person, !person && m == CallerModule(caller): + return nil + } + } + return fmt.Errorf("%s: refused for %s: not on the holder's caller list for %s", verb, caller, verb) +} + +// CallerModule is the module of a proved module caller, which the runtime stamps `.`. +func CallerModule(proved string) string { + if i := strings.LastIndexByte(proved, '.'); i >= 0 { + return proved[i+1:] + } + return proved +} diff --git a/modules/distribution/internal/holder/front.go b/modules/distribution/internal/holder/front.go new file mode 100644 index 000000000..eb42ae139 --- /dev/null +++ b/modules/distribution/internal/holder/front.go @@ -0,0 +1,101 @@ +package holder + +import ( + "crypto/subtle" + "fmt" + "net/http" + "net/http/httputil" + "net/url" + "strings" + "sync/atomic" +) + +// Front is the store's port, served by distribution's program in front of the registry container, which listens on its +// node's loopback only (design 51 *No build module holds a registry's write credential*): reads (the OCI distribution +// protocol's GET and HEAD) stay open to every node's container runtime; a write reaches the store only from +// distribution's own process, which writes to the loopback directly and never through the front. While the direct-write +// window is open, a build module's old build still pushes through the front as today, but never to a tag the seat holds +// published. +type Front struct { + // Store is the registry container's loopback address. + Store *url.URL + // Window says whether the direct-write window is open. + Window *atomic.Bool + // Published says whether name holds a tag published through the seat. + Published func(name, tag string) (bool, error) + // Bootstrap is the bootstrap's one-off credential for the holders' first images (design 51 *On a fresh mesh*): + // read on each write, never kept, so the file's removal (BootstrapFile) refuses the next write that carries it. + Bootstrap func() (token string, ok bool) + Logf func(string, ...any) + proxy *httputil.ReverseProxy +} + +// ServeHTTP answers one request of the store's port. +func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if f.proxy == nil { + f.proxy = httputil.NewSingleHostReverseProxy(f.Store) + } + switch r.Method { + case http.MethodGet, http.MethodHead: + f.proxy.ServeHTTP(w, r) + return + } + bootstrap := false + if f.Bootstrap != nil { + if token, ok := f.Bootstrap(); ok && token != "" { + got, has := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") + bootstrap = has && subtle.ConstantTimeCompare([]byte(got), []byte(token)) == 1 + } + } + if !bootstrap && (f.Window == nil || !f.Window.Load()) { + refuse(w, http.StatusForbidden, "DENIED", "the store takes writes from distribution's own process alone: publish through the seat's publish") + return + } + if name, tag, ok := manifestPut(r); ok && f.Published != nil { + held, err := f.Published(name, tag) + if err != nil { + refuse(w, http.StatusServiceUnavailable, "UNAVAILABLE", "whether "+name+":"+tag+" is published cannot be read, so nothing is written: "+err.Error()) + return + } + if held { + refuse(w, http.StatusForbidden, "DENIED", fmt.Sprintf("%s:%s is published through the seat; a direct write cannot overwrite it: publish through the seat's publish", name, tag)) + return + } + } + if bootstrap { + if f.Logf != nil { + f.Logf("a write %s %s from %s with the bootstrap's one-off credential", r.Method, r.URL.Path, r.RemoteAddr) + } + } else if f.Logf != nil { + f.Logf("a direct write %s %s from %s, accepted while the direct-write window is open", r.Method, r.URL.Path, r.RemoteAddr) + } + f.proxy.ServeHTTP(w, r) +} + +// BootstrapFile is where the node-engine's bootstrap gives distribution its one-off credential, in distribution's +// state directory (0600, distribution's account): one line, the token. The tools bundle removes it when it answers its +// first publish, so the holders serve publish and the credential is gone (RemoveBootstrap). +const BootstrapFile = "/var/lib/mesh/registry/bootstrap-push" + +// manifestPut is a PUT of /v2//manifests/, a tag and not a digest. +func manifestPut(r *http.Request) (name, tag string, ok bool) { + if r.Method != http.MethodPut { + return "", "", false + } + p := strings.TrimPrefix(r.URL.Path, "/v2/") + i := strings.LastIndex(p, "/manifests/") + if i <= 0 { + return "", "", false + } + name, tag = p[:i], p[i+len("/manifests/"):] + if strings.HasPrefix(tag, "sha256:") || tag == "" { + return "", "", false + } + return name, tag, true +} + +func refuse(w http.ResponseWriter, code int, kind, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + fmt.Fprintf(w, `{"errors":[{"code":%q,"message":%q}]}`, kind, msg) +} diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go new file mode 100644 index 000000000..2b1a68245 --- /dev/null +++ b/modules/distribution/internal/holder/holder_test.go @@ -0,0 +1,386 @@ +package holder + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" +) + +type memObjects map[string][]byte + +func (m memObjects) Put(n string, d []byte) error { m[n] = append([]byte(nil), d...); return nil } +func (m memObjects) Get(n string) ([]byte, error) { + if d, ok := m[n]; ok { + return d, nil + } + return nil, errors.New("no object " + n) +} +func (m memObjects) Delete(n string) error { delete(m, n); return nil } +func (m memObjects) Used() (uint64, error) { + var u uint64 + for _, d := range m { + u += uint64(len(d)) + } + return u, nil +} + +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) Delete(id string) error { delete(m, id); return nil } +func (m memTransfers) List() ([]Transfer, error) { + var out []Transfer + for _, t := range m { + out = append(out, t) + } + return out, nil +} + +type fakeBuilder struct { + ok bool + err error +} + +func (b fakeBuilder) Recorded(string, string, string) (bool, string, error) { + return b.ok, "not numbered", b.err +} + +type fakeStore struct { + has map[string]bool + put []map[string][]byte +} + +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) { + s.put = append(s.put, parts) + return "registry.mesh/" + t.Name + "@" + t.Digest, nil +} +func (s *fakeStore) Held(string, string, string) (string, bool, error) { return "", false, nil } + +type said struct { + seat, event string + body Event +} + +type bed struct { + h *Holder + v *Verbs + store *fakeStore + said []said + findings map[string]bool + now time.Time + cat *fakeCatalog +} + +type fakeCatalog struct { + tags map[string]string // name:version -> digest + removed []string +} + +func (c *fakeCatalog) Versions(name string) ([]string, error) { + var out []string + for k := range c.tags { + if n, v, _ := strings.Cut(k, ":"); n == name { + out = append(out, v) + } + } + return out, nil +} +func (c *fakeCatalog) Digest(name, version string) (string, error) { + return c.tags[name+":"+version], nil +} +func (c *fakeCatalog) Remove(name, version string) error { + c.removed = append(c.removed, name+":"+version) + delete(c.tags, name+":"+version) + return nil +} +func (c *fakeCatalog) RemoveDigest(name, digest string) error { + c.removed = append(c.removed, name+"@"+digest) + return nil +} +func (c *fakeCatalog) Mirror(source, digest string) (string, error) { + return "registry.mesh/upstream/" + source + "@" + digest, nil +} + +type memHolds map[string]Hold + +func (m memHolds) Get(k string) (Hold, bool, error) { h, ok := m[k]; return h, ok, nil } +func (m memHolds) Put(h Hold) error { m[h.Key()] = h; return nil } +func (m memHolds) Delete(k string) error { delete(m, k); return nil } +func (m memHolds) List() ([]Hold, error) { + var out []Hold + for _, h := range m { + out = append(out, h) + } + return out, nil +} + +func newBed(builderOK bool) *bed { + b := &bed{store: &fakeStore{has: map[string]bool{}}, findings: map[string]bool{}, now: time.Unix(1000, 0), cat: &fakeCatalog{tags: map[string]string{}}} + b.h = &Holder{Callers: DefaultCallers(), Objects: memObjects{}, Transfers: memTransfers{}, Builder: fakeBuilder{ok: builderOK}, + Store: b.store, Now: func() time.Time { return b.now }, + Say: func(seat, event string, body Event) error { + b.said = append(b.said, said{seat, event, body}) + return nil + }, + Finding: func(kind, subject, _ string, raised bool) { b.findings[kind+" "+subject] = raised }} + b.v = &Verbs{Holder: b.h, Catalog: b.cat, Holds: memHolds{}} + return b +} + +func digestOf(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) } + +const builder = "ace.mesh-build-go" + +// TestTheHolderOfTheTwoImageSeats holds design 51's holder cases at the unit level (CAT-D1, D2, D3, D3b, D4, D4b, D5): +// one table or case per rule, the caller list and the store front's. +func TestTheHolderOfTheTwoImageSeats(t *testing.T) { + blob := []byte(strings.Repeat("a", 3000)) + part := Part{Digest: digestOf(blob), Size: int64(len(blob))} + req := func(id string) PublishRequest { + return PublishRequest{Transfer: id, Name: "m/program", Version: "0.1.4", Kind: KindBundle, Digest: part.Digest, Parts: []Part{part}} + } + + t.Run("a transfer in parts, cut and resumed, hashed, numbered by the builder, published", func(t *testing.T) { + b := newBed(true) + a, err := b.h.Publish(ArtifactSeat, builder, req("t1")) + if err != nil || !reflect.DeepEqual(a.Missing, []string{part.Digest}) { + t.Fatalf("publish = %+v, %v", a, err) + } + if _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 0, Data: blob[:1000]}); err != nil { + t.Fatal(err) + } + // Cut: the build module opens again and is told what is still missing; its next chunk goes on from there. + if a, _ := b.h.Publish(ArtifactSeat, builder, req("t1")); !reflect.DeepEqual(a.Missing, []string{part.Digest}) { + t.Errorf("resumed publish = %+v", a) + } + if _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 0, Data: blob[:1000]}); err == nil { + t.Error("a chunk sent twice was taken: the transfer goes on from the chunks held") + } + a, err = b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 1, Last: true, Data: blob[1000:]}) + if err != nil || !a.Ended || len(b.said) != 1 || b.said[0].event != EventPublished || b.said[0].body.Reference == "" { + t.Fatalf("last chunk = %+v, %v; said %+v", a, err, b.said) + } + if len(b.h.Objects.(memObjects)) != 0 || len(b.h.Transfers.(memTransfers)) != 0 { + t.Error("the transfer's chunks or key stayed after its end") + } + }) + t.Run("refusals", func(t *testing.T) { + for _, c := range []struct { + name string + do func(b *bed) error + want string + }{ + {"a caller not on the list", func(b *bed) error { _, err := b.h.Publish(ArtifactSeat, "ace.mesh-provisioner", req("t")); return err }, "not on the holder's caller list"}, + {"an unproven caller", func(b *bed) error { _, err := b.h.Publish(ArtifactSeat, Unproven, req("t")); return err }, "no caller the tool runner proved"}, + {"a part from another caller than the opener", func(b *bed) error { + _, _ = b.h.Publish(ArtifactSeat, builder, req("t")) + _, err := b.h.PublishPart(ArtifactSeat, "ace.mesh-build-node", PartRequest{Transfer: "t", Digest: part.Digest, Last: true, Data: blob}) + return err + }, "opened by another caller"}, + {"a part whose bytes do not match its digest", func(b *bed) error { + _, _ = b.h.Publish(ArtifactSeat, builder, req("t")) + _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t", Digest: part.Digest, Last: true, Data: []byte("other")}) + return err + }, "hash to"}, + {"a chunk over 1 MiB", func(b *bed) error { + _, _ = b.h.Publish(ArtifactSeat, builder, req("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 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 + _, err := b.h.Publish(ArtifactSeat, builder, req("t")) + return err + }, "the builder did not number it"}, + {"an image on the artifact seat", func(b *bed) error { + r := req("t") + r.Kind = KindImage + _, err := b.h.Publish(ArtifactSeat, builder, r) + return err + }, "holds bundles and archives"}, + {"remove of a version with no tag", func(b *bed) error { + return b.v.Remove(ArtifactSeat, "novox.mesh-delivery-builder", RemoveRequest{Name: "m/program", Version: "0.1.4", Reason: "branch ended"}) + }, "has no tag"}, + {"a retention removal from another caller than the sweep", func(b *bed) error { + return b.v.Remove(ArtifactSeat, "novox.mesh-delivery-builder", RemoveRequest{Name: "m/program", Digest: part.Digest, Reason: ReasonRetention}) + }, "artifact sweep's alone"}, + {"a retention removal of a held archive", func(b *bed) error { + _ = b.v.HoldOne(ArtifactSeat, "novox.artifact-store-tools", HoldRequest{Name: "m/a", Kind: KindArchive, Digest: part.Digest}) + return b.v.Remove(ArtifactSeat, "novox.artifact-store-tools", RemoveRequest{Name: "m/a", Digest: part.Digest, Reason: ReasonRetention}) + }, "is held"}, + {"a hold of an image", func(b *bed) error { + return b.v.HoldOne(ImageSeat, "novox.artifact-store-tools", HoldRequest{Name: "m/i", Kind: KindImage, Digest: part.Digest}) + }, "never held"}, + {"a hold from another caller", func(b *bed) error { + return b.v.HoldOne(ArtifactSeat, "novox.mesh-build-go", HoldRequest{Name: "m/a", Kind: KindArchive, Digest: part.Digest}) + }, "not on the holder's caller list"}, + {"a release from a caller not on the list", func(b *bed) error { + return b.v.Release(ArtifactSeat, "novox.mesh-build-go", HoldRequest{Name: "m/a", Digest: part.Digest}) + }, "not on the holder's caller list"}, + {"a release in distribution's own name", func(b *bed) error { + return b.v.Release(ArtifactSeat, "novox.distribution", HoldRequest{Name: "m/a", Digest: part.Digest}) + }, "in its own name"}, + {"a mirror by no digest", func(b *bed) error { + _, err := b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "docker.io/library/nats:2"}) + return err + }, "pinned digest"}, + } { + b := newBed(true) + if err := c.do(b); err == nil || !strings.Contains(err.Error(), c.want) { + t.Errorf("%s: err = %v, want one saying %q", c.name, err, c.want) + } + } + if _, err := ParseCallers(`{"release":["distribution"]}`); err == nil { + t.Error("a caller list putting distribution on release's list is taken") + } + }) + t.Run("only the missing layers are sent, and an idle transfer is dropped", func(t *testing.T) { + b := newBed(true) + other := []byte("layer two") + r := req("t2") + r.Kind, r.Parts = KindImage, []Part{part, {Digest: digestOf(other), Size: int64(len(other))}} + b.store.has[part.Digest] = true + a, err := b.h.Publish(ImageSeat, builder, r) + if err != nil || !reflect.DeepEqual(a.Missing, []string{digestOf(other)}) { + t.Fatalf("publish = %+v, %v; want only the layer the store lacks", a, err) + } + b.now = b.now.Add(TransferIdle) + if dropped, _ := b.h.DropIdle(); !reflect.DeepEqual(dropped, []string{"t2"}) { + t.Errorf("DropIdle = %v, want t2 after 24 hours with no part", dropped) + } + }) + t.Run("incoming's finding at 80 %, cleared below", func(t *testing.T) { + b := newBed(true) + b.h.Cap = 3500 + _, _ = b.h.Publish(ArtifactSeat, builder, req("t3")) + _, _ = b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t3", Digest: part.Digest, Data: blob[:2800]}) + if !b.findings["incoming-full incoming"] { + t.Error("2800 of 3500 bytes wrote no finding") + } + _, _ = b.h.Drop("t3") + if b.findings["incoming-full incoming"] { + t.Error("the finding stayed after the transfer was dropped") + } + }) + t.Run("the sweep's removal, hold and release", func(t *testing.T) { + b := newBed(true) + sweep := "novox.artifact-store-tools" + if err := b.v.Remove(ArtifactSeat, sweep, RemoveRequest{Name: "m/a", Digest: part.Digest, Kind: KindArchive, Reason: ReasonRetention}); err != nil { + t.Fatal(err) + } + if err := b.v.HoldOne(ArtifactSeat, sweep, HoldRequest{Name: "m/b", Kind: KindArchive, Digest: part.Digest}); err != nil { + t.Fatal(err) + } + if err := b.v.Release(ArtifactSeat, "person.jochen", HoldRequest{Name: "m/b", Digest: part.Digest, Why: "by hand"}); err != nil { + t.Errorf("the operator's session releasing through registry_holds: %v", err) + } + var events []string + for _, s := range b.said { + events = append(events, s.event+" "+s.body.By) + } + if want := []string{"removed " + sweep, "held " + sweep, "released person.jochen"}; !reflect.DeepEqual(events, want) { + t.Errorf("said %v, want %v", events, want) + } + b.cat.tags["m/p:0.1.5-feat-x.1"] = part.Digest + if err := b.v.Remove(ArtifactSeat, "novox.mesh-delivery-builder", RemoveRequest{Name: "m/p", Version: "0.1.5-feat-x.1", Reason: "branch merged"}); err != nil || b.cat.tags["m/p:0.1.5-feat-x.1"] != "" { + t.Errorf("the builder's removal of a tagged version: %v", err) + } + }) + t.Run("the store's port: reads open, writes from distribution alone, and the direct-write window", func(t *testing.T) { + store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusCreated) })) + defer store.Close() + u, _ := url.Parse(store.URL) + var open atomic.Bool + token, kept := "s3cret", true + f := &Front{Store: u, Window: &open, Published: func(name, tag string) (bool, error) { return name == "m/i" && tag == "0.1.3", nil }, + Bootstrap: func() (string, bool) { return token, kept }} + auth := "" + do := func(method, path string) int { + rec := httptest.NewRecorder() + req := httptest.NewRequest(method, path, nil) + if auth != "" { + req.Header.Set("Authorization", "Bearer "+auth) + } + f.ServeHTTP(rec, req) + return rec.Code + } + for _, c := range []struct { + open bool + method, path string + want int + }{ + {false, http.MethodGet, "/v2/m/i/manifests/0.1.3", http.StatusCreated}, + {false, http.MethodHead, "/v2/m/i/blobs/sha256:aa", http.StatusCreated}, + {false, http.MethodPut, "/v2/m/i/manifests/0.1.9", http.StatusForbidden}, + {false, http.MethodPost, "/v2/m/i/blobs/uploads/", http.StatusForbidden}, + {true, http.MethodPost, "/v2/m/i/blobs/uploads/", http.StatusCreated}, + {true, http.MethodPut, "/v2/m/i/manifests/0.1.9", http.StatusCreated}, + {true, http.MethodPut, "/v2/m/i/manifests/0.1.3", http.StatusForbidden}, // published through the seat + } { + open.Store(c.open) + if got := do(c.method, c.path); got != c.want { + t.Errorf("window open %v: %s %s = %d, want %d", c.open, c.method, c.path, got, c.want) + } + } + // The bootstrap's one-off credential writes with the window closed, never over a published tag, and not once + // its file is gone; a wrong token writes nothing. + open.Store(false) + for _, c := range []struct { + auth string + kept bool + path string + want int + }{ + {"s3cret", true, "/v2/gitea/program/manifests/0.1.0", http.StatusCreated}, + {"s3cret", true, "/v2/m/i/manifests/0.1.3", http.StatusForbidden}, + {"wrong", true, "/v2/gitea/program/manifests/0.1.0", http.StatusForbidden}, + {"s3cret", false, "/v2/gitea/program/manifests/0.1.0", http.StatusForbidden}, + } { + auth, kept = c.auth, c.kept + if got := do(http.MethodPut, c.path); got != c.want { + t.Errorf("bootstrap %q kept %v: PUT %s = %d, want %d", c.auth, c.kept, c.path, got, c.want) + } + } + }) + t.Run("the windows", func(t *testing.T) { + nodes := []NodeRead{ + {Node: "ace", Serves: true, Stated: []string{NewNamesCapability}, NamesArtifact: true, BuildModules: map[string]bool{"mesh-build-go": true}}, + {Node: "g14", NamesArtifact: true, BuildModules: map[string]bool{"mesh-build-go": false}}, // away, old node-engine + {Node: "novox", Serves: true, Stated: nil, NamesArtifact: false}, + } + if got := HoldingOldName(nodes, nil); !reflect.DeepEqual(got, []string{"g14"}) { + t.Errorf("old name held by %v, want g14 (no manifest-reads: it states nothing)", got) + } + if got := HoldingDirectWrites(nodes, nil); !reflect.DeepEqual(got, []string{"g14"}) { + t.Errorf("direct writes held by %v, want g14", got) + } + t0 := time.Unix(0, 0) + w, closed, _ := Window{Name: OldNameWindow}.Step([]string{"g14"}, t0) + if closed != nil || !w.Open { + t.Fatalf("Step = %+v, %+v", w, closed) + } + if _, _, overdue := w.Step([]string{"g14"}, t0.Add(WindowFindingAfter)); !reflect.DeepEqual(overdue, []string{"g14"}) { + t.Errorf("seven days: overdue %v", overdue) + } + w, _ = w.Drop("g14", "person.jochen", "retired", t0) + w, closed, _ = w.Step(HoldingOldName(nodes, w.Dropped), t0.Add(time.Hour)) + if closed == nil || closed.Window != OldNameWindow || !reflect.DeepEqual(closed.LastHolding, []string{"g14"}) || len(closed.Dropped) != 1 { + t.Errorf("closing = %+v", closed) + } + if _, again, _ := w.Step([]string{"g14"}, t0.Add(2*time.Hour)); again != nil { + t.Error("a closed window closed again") + } + }) +} diff --git a/modules/distribution/internal/holder/oci.go b/modules/distribution/internal/holder/oci.go new file mode 100644 index 000000000..10bfa5f2b --- /dev/null +++ b/modules/distribution/internal/holder/oci.go @@ -0,0 +1,262 @@ +package holder + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "sort" + "strings" +) + +// OCI is the artifact store as the OCI distribution protocol answers it on its node's loopback, written by +// distribution's own process alone: the Store and the Catalog the verbs use. Registry is the reach the seat answers in a +// reference (`//@sha256:`, ADR 0605 §9). +type OCI struct { + Base *url.URL + Registry string + Client *http.Client +} + +const ( + mediaManifest = "application/vnd.oci.image.manifest.v1+json" + mediaEmptyConfig = "application/vnd.oci.empty.v1+json" + mediaLayer = "application/vnd.oci.image.layer.v1.tar" + emptyConfig = "{}" +) + +func (o *OCI) client() *http.Client { + if o.Client != nil { + return o.Client + } + return http.DefaultClient +} + +func (o *OCI) url(path string) string { return strings.TrimSuffix(o.Base.String(), "/") + path } + +func (o *OCI) do(method, path, contentType string, body []byte, accept ...string) (*http.Response, []byte, error) { + req, err := http.NewRequest(method, o.url(path), bytes.NewReader(body)) + if err != nil { + return nil, nil, err + } + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + for _, a := range accept { + req.Header.Add("Accept", a) + } + resp, err := o.client().Do(req) + if err != nil { + return nil, nil, err + } + defer resp.Body.Close() + b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) + return resp, b, nil +} + +// Has is whether name holds a blob. +func (o *OCI) Has(name, digest string) (bool, error) { + resp, _, err := o.do(http.MethodHead, "/v2/"+name+"/blobs/"+digest, "", nil) + if err != nil { + return false, err + } + return resp.StatusCode == http.StatusOK, nil +} + +func (o *OCI) putBlob(name, digest string, data []byte) error { + resp, b, err := o.do(http.MethodPost, "/v2/"+name+"/blobs/uploads/", "", nil) + if err != nil { + return err + } + if resp.StatusCode != http.StatusAccepted { + return fmt.Errorf("the store did not open an upload of %s (%d): %s", digest, resp.StatusCode, b) + } + loc, err := url.Parse(resp.Header.Get("Location")) + if err != nil { + return err + } + q := loc.Query() + q.Set("digest", digest) + loc.RawQuery = q.Encode() + path := loc.Path + if loc.RawQuery != "" { + path += "?" + loc.RawQuery + } + resp, b, err = o.do(http.MethodPut, path, "application/octet-stream", data) + if err != nil { + return err + } + if resp.StatusCode != http.StatusCreated { + return fmt.Errorf("the store did not take %s (%d): %s", digest, resp.StatusCode, b) + } + return nil +} + +func (o *OCI) putManifest(name, ref, mediaType string, data []byte) (string, error) { + resp, b, err := o.do(http.MethodPut, "/v2/"+name+"/manifests/"+ref, mediaType, data) + if err != nil { + return "", err + } + if resp.StatusCode != http.StatusCreated { + return "", fmt.Errorf("the store did not take the manifest %s:%s (%d): %s", name, ref, resp.StatusCode, b) + } + return "sha256:" + sum(data), nil +} + +// 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) { + var manifest *Part + digests := make([]string, 0, len(parts)) + for d := range parts { + digests = append(digests, d) + } + sort.Strings(digests) + for i, p := range t.Parts { + if strings.Contains(p.MediaType, "manifest") { + manifest = &t.Parts[i] + continue + } + if data, ok := parts[p.Digest]; ok { + if err := o.putBlob(t.Name, p.Digest, data); err != nil { + return "", err + } + } + } + ref := t.Version + switch t.Kind { + case KindImage, KindUpstream: + if manifest == nil { + return "", fmt.Errorf("an image's publish names no manifest part") + } + data, ok := parts[manifest.Digest] + if !ok { + // Held already: fetch it to tag it under the version. + resp, b, err := o.do(http.MethodGet, "/v2/"+t.Name+"/manifests/"+manifest.Digest, "", nil, manifest.MediaType) + if err != nil || resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("the held manifest %s cannot be read: %v", manifest.Digest, err) + } + data = b + } + if ref == "" { + ref = manifest.Digest + } + d, err := o.putManifest(t.Name, ref, manifest.MediaType, data) + if err != nil { + return "", err + } + return o.Registry + "/" + t.Name + "@" + d, nil + default: + if err := o.putBlob(t.Name, "sha256:"+sum([]byte(emptyConfig)), []byte(emptyConfig)); err != nil { + return "", err + } + size := int64(0) + for _, p := range t.Parts { + if p.Digest == t.Digest { + size = p.Size + } + } + m, _ := json.Marshal(map[string]any{ + "schemaVersion": 2, "mediaType": mediaManifest, + "config": map[string]any{"mediaType": mediaEmptyConfig, "digest": "sha256:" + sum([]byte(emptyConfig)), "size": len(emptyConfig)}, + "layers": []map[string]any{{"mediaType": mediaLayer, "digest": t.Digest, "size": size}}, + }) + if ref == "" { + ref = "sha256:" + sum(m) + } + if _, err := o.putManifest(t.Name, ref, mediaManifest, m); err != nil { + return "", err + } + return o.Registry + "/" + t.Name + "@" + t.Digest, nil + } +} + +// Held is the reference of name at version when the store holds digest under it. +func (o *OCI) Held(name, version, digest string) (string, bool, error) { + d, err := o.Digest(name, version) + if err != nil || d == "" { + return "", false, err + } + if d != digest { + return "", false, nil + } + return o.Registry + "/" + name + "@" + d, true, nil +} + +// Versions are name's tags. +func (o *OCI) Versions(name string) ([]string, error) { + resp, b, err := o.do(http.MethodGet, "/v2/"+name+"/tags/list", "", nil) + if err != nil { + return nil, err + } + if resp.StatusCode == http.StatusNotFound { + return []string{}, nil + } + var out struct { + Tags []string `json:"tags"` + } + if resp.StatusCode != http.StatusOK || json.Unmarshal(b, &out) != nil { + return nil, fmt.Errorf("the store's tags of %s (%d): %s", name, resp.StatusCode, b) + } + sort.Strings(out.Tags) + return out.Tags, nil +} + +// Digest is the manifest digest name holds at version, empty when none. +func (o *OCI) Digest(name, version string) (string, error) { + resp, _, err := o.do(http.MethodHead, "/v2/"+name+"/manifests/"+version, "", nil, mediaManifest, + "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.v2+json") + if err != nil { + return "", err + } + if resp.StatusCode == http.StatusNotFound { + return "", nil + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("the store answered %d for %s:%s", resp.StatusCode, name, version) + } + return resp.Header.Get("Docker-Content-Digest"), nil +} + +// Remove deletes the manifest name holds at version. +func (o *OCI) Remove(name, version string) error { + d, err := o.Digest(name, version) + if err != nil || d == "" { + return err + } + return o.RemoveDigest(name, d) +} + +// RemoveDigest deletes a manifest by its digest; the store's nightly collector frees its blobs. +func (o *OCI) RemoveDigest(name, digest string) error { + resp, b, err := o.do(http.MethodDelete, "/v2/"+name+"/manifests/"+digest, "", nil) + if err != nil { + return err + } + if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusNotFound { + return fmt.Errorf("the store did not delete %s@%s (%d): %s", name, digest, resp.StatusCode, b) + } + return nil +} + +// Mirror is not done by the store itself: mesh-build-image sends an upstream image's parts through publish of kind +// upstream, named `upstream//`; mirror answers where a held copy is, and refuses one the store lacks so +// the build module publishes it. +func (o *OCI) Mirror(source, digest string) (string, error) { + name := "upstream/" + source + if i := strings.LastIndexAny(name, ":@"); i > strings.LastIndex(name, "/") { + name = name[:i] + } + resp, _, err := o.do(http.MethodHead, "/v2/"+name+"/manifests/"+digest, "", nil, "application/vnd.oci.image.index.v1+json", + mediaManifest, "application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.docker.distribution.manifest.v2+json") + if err != nil { + return "", err + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("the store holds no copy of %s@%s: publish it with kind upstream under %s", source, digest, name) + } + return o.Registry + "/" + name + "@" + digest, nil +} diff --git a/modules/distribution/internal/holder/transfer.go b/modules/distribution/internal/holder/transfer.go new file mode 100644 index 000000000..97bd0c67a --- /dev/null +++ b/modules/distribution/internal/holder/transfer.go @@ -0,0 +1,448 @@ +package holder + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "slices" + "sort" + "strings" + "time" +) + +// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, the store 20 GiB, a transfer +// with no part for 24 hours is dropped, and past 80 % of the cap the holder writes a finding. +const ( + MaxPart = 1 << 20 + IncomingCap = 20 << 30 + TransferIdle = 24 * time.Hour + IncomingFullPct = 80 + // IncomingObjects and TransfersState are the object store and the state distribution asks as bus state. + IncomingObjects = "incoming" + TransfersState = "transfers" +) + +// The kinds a seat holds. +const ( + KindImage = "image" + KindBundle = "bundle" + KindArchive = "archive" + KindUpstream = "upstream" +) + +// Part is one file of a publish: an image's layer, config or manifest, a bundle's or an archive's blob, by its digest. +type Part struct { + Digest string `json:"digest"` + Size int64 `json:"size"` + // MediaType says what the part is to the store; an image's manifest names its own. + MediaType string `json:"media_type,omitempty"` +} + +// PublishRequest is the body of `publish` (ADR 0611 §1). +type PublishRequest struct { + Transfer string `json:"transfer"` + Name string `json:"name"` + Version string `json:"version,omitempty"` + Tag string `json:"tag,omitempty"` + Kind string `json:"kind"` + Fingerprint string `json:"fingerprint,omitempty"` + Commit string `json:"commit,omitempty"` + // Digest is the digest the build recorded: an image's manifest, a bundle's or an archive's blob. + Digest string `json:"digest"` + Parts []Part `json:"parts"` +} + +// PublishAnswer is what `publish` answers: the parts the holder lacks, which the build module sends; none when it +// holds them all and the publish ended (Published names how). +type PublishAnswer struct { + Transfer string `json:"transfer"` + Missing []string `json:"missing"` + // Ended says the publish ended at once: everything was held, `published` or `refused` is said. + Ended bool `json:"ended,omitempty"` +} + +// PartRequest is the body of `publish-part` (ADR 0611 §2): one chunk of one part. +type PartRequest struct { + Transfer string `json:"transfer"` + Digest string `json:"digest"` + Sequence int `json:"sequence"` + Last bool `json:"last"` + Data []byte `json:"data"` +} + +// Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming. +type Transfer 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"` +} + +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 +} + +// 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. +type Event struct { + Name string `json:"name"` + Version string `json:"version,omitempty"` + Kind string `json:"kind"` + Tag string `json:"tag,omitempty"` + Digest string `json:"digest"` + Fingerprint string `json:"fingerprint,omitempty"` + Commit string `json:"commit,omitempty"` + Holder string `json:"holder"` + Transfer string `json:"transfer,omitempty"` + // Reference is the seat's answer: `//@sha256:` (ADR 0605 §9). + Reference string `json:"reference,omitempty"` + Reason string `json:"reason,omitempty"` + // By is who asked a removal, hold or release: the caller the tool runner proved. + By string `json:"by,omitempty"` +} + +// The events of the seats. +const ( + EventPublished = "published" + EventRefused = "refused" + EventRemoved = "removed" + EventHeld = "held" + EventReleased = "released" +) + +// 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) +} + +// 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) +} + +// Builder is the builder's tool builder_versions, asked before anything is stored (ADR 0605 §4). +type Builder interface { + Recorded(name, version, digest string) (ok bool, why string, err error) +} + +// Store is the artifact store behind the loopback, written by distribution's own process alone. +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) + // 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) +} + +// Say says an event on a seat's family `.events`. +type Say func(seat, event string, body Event) error + +// Finding writes (raised true) or clears distribution's finding of a kind for a subject. +type Finding func(kind, subject, words string, raised bool) + +// Holder answers the verbs. +type Holder struct { + Callers Callers + Objects Objects + Transfers TransferState + Builder Builder + Store Store + Say Say + Finding Finding + Now func() time.Time + // Cap is incoming's cap; IncomingCap when zero. + Cap uint64 +} + +func (h *Holder) now() time.Time { + if h.Now != nil { + return h.Now().UTC() + } + return time.Now().UTC() +} + +func (h *Holder) cap() uint64 { + if h.Cap > 0 { + return h.Cap + } + return IncomingCap +} + +// 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 { + return PublishAnswer{}, err + } + 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 { + return PublishAnswer{}, fmt.Errorf("publish: the store cannot say whether it holds %s@%s: %w", r.Name, r.Version, err) + } else if ok { + return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.Say(seat, EventPublished, h.eventOf(seat, r, ref, "")) + } + } + used, err := h.Objects.Used() + if err != nil { + return PublishAnswer{}, fmt.Errorf("publish: incoming's fill cannot be read: %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 +} + +// PublishPart is `publish-part`: one chunk, from the caller that opened the transfer alone. +func (h *Holder) PublishPart(seat, caller string, r PartRequest) (PublishAnswer, error) { + 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 + } + 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) + if err != nil { + return refuse("the store did not take it: " + err.Error()) + } + h.Finding("publish-refused", t.Name+"@"+t.Version, "", false) + return h.Say(t.Seat, EventPublished, h.eventOf(t.Seat, t.PublishRequest, ref, "")) +} + +func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Event { + return Event{Name: r.Name, Version: r.Version, Kind: r.Kind, Tag: r.Tag, Digest: r.Digest, Fingerprint: r.Fingerprint, + 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 +} + +// 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 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: + return fmt.Errorf("publish: %s holds images, not a %s", seat, r.Kind) + 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 +} diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go new file mode 100644 index 000000000..4406e7952 --- /dev/null +++ b/modules/distribution/internal/holder/verbs.go @@ -0,0 +1,235 @@ +package holder + +import ( + "errors" + "fmt" + "slices" + "strings" + "time" +) + +// Catalog is what the store answers of what it holds. +type Catalog interface { + // Versions are the versions (tags) of name, sorted. + Versions(name string) ([]string, error) + // Digest is the digest name holds at version, empty when none. + Digest(name, version string) (string, error) + // Remove deletes name at version (its manifest); the store's collector frees its blobs. + Remove(name, version string) error + // RemoveDigest deletes an untagged manifest of name by its digest (retention). + RemoveDigest(name, digest string) error + // Mirror copies an outside image, pinned by digest, into upstream//, answering the reference. + Mirror(source, digest string) (string, error) +} + +// HoldState is distribution's own state holds (bus state, template state; ADR 0617 §2). +type HoldState interface { + Get(key string) (Hold, bool, error) + Put(h Hold) error + Delete(key string) error + List() ([]Hold, error) +} + +// Hold keeps one untagged bundle, archive or copied base from retention's removal. +type Hold struct { + Seat string `json:"seat"` + Name string `json:"name"` + Kind string `json:"kind"` + Digest string `json:"digest"` + By string `json:"by"` + Why string `json:"why,omitempty"` + Since time.Time `json:"since"` +} + +// Key is the hold's key in the state. +func (h Hold) Key() string { + return h.Seat + "." + strings.ReplaceAll(strings.ReplaceAll(h.Name, ".", "="), "@", "=") + "." + strings.TrimPrefix(h.Digest, "sha256:") +} + +// Verbs answers the verbs other than the publish: versions, remove, hold, release, tags and mirror. +type Verbs struct { + *Holder + Catalog Catalog + Holds HoldState +} + +// VersionsAnswer is `versions`: the versions the seat holds of name. +type VersionsAnswer struct { + Seat string `json:"seat"` + Name string `json:"name"` + Versions []string `json:"versions"` +} + +// Versions is `versions`. +func (v *Verbs) Versions(seat, caller, name string) (VersionsAnswer, error) { + if err := v.Callers.Allowed(VerbVersions, caller); err != nil { + return VersionsAnswer{}, err + } + if strings.TrimSpace(name) == "" { + return VersionsAnswer{}, errors.New("versions: no name") + } + got, err := v.Catalog.Versions(name) + if err != nil { + return VersionsAnswer{}, fmt.Errorf("versions: the store cannot list %s: %w", name, err) + } + if got == nil { + got = []string{} + } + return VersionsAnswer{Seat: seat, Name: name, Versions: got}, nil +} + +// RemoveRequest is `remove`: a tagged version, or with the reason retention an untagged one by its digest. +type RemoveRequest struct { + Name string `json:"name"` + Version string `json:"version,omitempty"` + Digest string `json:"digest,omitempty"` + Kind string `json:"kind,omitempty"` + Reason string `json:"reason"` +} + +// ReasonRetention is the reason of the sweep's removal of an untagged version (ADR 0610 §3). +const ReasonRetention = "retention" + +// Remove is `remove`: the builder removes a tagged version; the artifact sweep's holder an untagged bundle, archive or +// copied base with the reason retention, which a hold keeps. Every removal says `removed` with its reason. +func (v *Verbs) Remove(seat, caller string, r RemoveRequest) error { + if err := v.Callers.Allowed(VerbRemove, caller); err != nil { + return err + } + if strings.TrimSpace(r.Name) == "" || strings.TrimSpace(r.Reason) == "" { + return errors.New("remove: a name and a reason are required") + } + ev := Event{Name: r.Name, Version: r.Version, Kind: r.Kind, Digest: r.Digest, Holder: Self, Reason: r.Reason, By: caller} + switch { + case r.Reason == ReasonRetention: + if CallerModule(caller) != SweepHolder { + return fmt.Errorf("remove: refused for %s: a removal for retention is the artifact sweep's alone", caller) + } + if r.Digest == "" { + return errors.New("remove: a removal for retention names the digest") + } + if _, held, err := v.Holds.Get(Hold{Seat: seat, Name: r.Name, Digest: r.Digest}.Key()); err != nil { + return fmt.Errorf("remove: the holds cannot be read, so nothing is removed: %w", err) + } else if held { + return fmt.Errorf("remove: refused: %s@%s is held; release it first", r.Name, r.Digest) + } + if err := v.Catalog.RemoveDigest(r.Name, r.Digest); err != nil { + return fmt.Errorf("remove: the store did not remove %s@%s: %w", r.Name, r.Digest, err) + } + default: + if r.Version == "" || !strings.Contains(r.Version, "-") { + return fmt.Errorf("remove: refused: %s@%s has no tag; only a tagged version is removed, but for retention", r.Name, r.Version) + } + d, err := v.Catalog.Digest(r.Name, r.Version) + if err != nil { + return fmt.Errorf("remove: the store cannot read %s@%s: %w", r.Name, r.Version, err) + } + if d != "" { + if err := v.Catalog.Remove(r.Name, r.Version); err != nil { + return fmt.Errorf("remove: the store did not remove %s@%s: %w", r.Name, r.Version, err) + } + } + ev.Digest = d // a version already gone is said removed all the same, so the builder's retry ends + } + return v.Say(seat, EventRemoved, ev) +} + +// HoldRequest is `hold` and `release`. +type HoldRequest struct { + Name string `json:"name"` + Kind string `json:"kind"` + Digest string `json:"digest"` + Why string `json:"why,omitempty"` +} + +// HoldOne is `hold`: the artifact sweep's holder alone; an image is never held (its own rules keep images). +func (v *Verbs) HoldOne(seat, caller string, r HoldRequest) error { + if err := v.Callers.Allowed(VerbHold, caller); err != nil { + return err + } + switch r.Kind { + case KindBundle, KindArchive, KindUpstream: + default: + return fmt.Errorf("hold: refused: a %s is never held; only an untagged bundle, archive or copied base", r.Kind) + } + if r.Name == "" || !strings.HasPrefix(r.Digest, "sha256:") { + return errors.New("hold: a name and a sha256 digest are required") + } + h := Hold{Seat: seat, Name: r.Name, Kind: r.Kind, Digest: r.Digest, By: caller, Why: r.Why, Since: v.now()} + if err := v.Holds.Put(h); err != nil { + return fmt.Errorf("hold: the holds did not take it: %w", err) + } + return v.Say(seat, EventHeld, Event{Name: r.Name, Kind: r.Kind, Digest: r.Digest, Holder: Self, By: caller}) +} + +// Release is `release`: the artifact sweep's holder, or the operator's session through registry_holds (a hand act, +// recorded by the tool); distribution's own account is never on its list. +func (v *Verbs) Release(seat, caller string, r HoldRequest) error { + if CallerModule(caller) == Self && !strings.HasPrefix(caller, "person.") { + return fmt.Errorf("release: refused for %s: distribution cannot release in its own name", caller) + } + if err := v.Callers.Allowed(VerbRelease, caller); err != nil { + return err + } + key := Hold{Seat: seat, Name: r.Name, Digest: r.Digest}.Key() + h, held, err := v.Holds.Get(key) + if err != nil { + return err + } + if !held { + return fmt.Errorf("release: %s@%s is not held on %s", r.Name, r.Digest, seat) + } + if err := v.Holds.Delete(key); err != nil { + return err + } + return v.Say(seat, EventReleased, Event{Name: h.Name, Kind: h.Kind, Digest: h.Digest, Holder: Self, By: caller, Reason: r.Why}) +} + +// Tags is `tags` on the image seat: the tags of an image. +func (v *Verbs) Tags(seat, caller, name string) (VersionsAnswer, error) { + if seat != ImageSeat { + return VersionsAnswer{}, fmt.Errorf("tags: %s serves no tags", seat) + } + if err := v.Callers.Allowed(VerbTags, caller); err != nil { + return VersionsAnswer{}, err + } + got, err := v.Catalog.Versions(name) + if err != nil { + return VersionsAnswer{}, err + } + return VersionsAnswer{Seat: seat, Name: name, Versions: got}, nil +} + +// MirrorRequest is `mirror`: an outside image by its pinned digest. +type MirrorRequest struct { + Source string `json:"source"` + Digest string `json:"digest"` +} + +// Mirror is `mirror` on the image seat: mesh-build-image alone; an image pinned by no digest is refused. +func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) { + if seat != ImageSeat { + return "", fmt.Errorf("mirror: %s serves no mirror", seat) + } + if err := v.Callers.Allowed(VerbMirror, caller); err != nil { + return "", err + } + if r.Source == "" || !strings.HasPrefix(r.Digest, "sha256:") { + return "", errors.New("mirror: refused: an upstream image is copied only by its pinned digest") + } + ref, err := v.Catalog.Mirror(r.Source, r.Digest) + if err != nil { + return "", fmt.Errorf("mirror: %s@%s was not copied: %w", r.Source, r.Digest, err) + } + return ref, v.Say(seat, EventPublished, Event{Name: "upstream/" + r.Source, Kind: KindUpstream, Digest: r.Digest, Holder: Self, Reference: ref, By: caller}) +} + +// HeldNow lists the holds, for registry_holds. +func (v *Verbs) HeldNow() ([]Hold, error) { + all, err := v.Holds.List() + if err != nil { + return nil, err + } + slices.SortFunc(all, func(a, b Hold) int { return strings.Compare(a.Key(), b.Key()) }) + return all, nil +} diff --git a/modules/distribution/internal/holder/window.go b/modules/distribution/internal/holder/window.go new file mode 100644 index 000000000..8b6268237 --- /dev/null +++ b/modules/distribution/internal/holder/window.go @@ -0,0 +1,148 @@ +package holder + +import ( + "fmt" + "slices" + "sort" + "time" +) + +// The two windows distribution keeps (design 51): the old seat name `mesh-artifact-store`, served until every node whose +// declaration names an artifact applied a build whose node-engine states it reads the new names (ADR 0610 §4, ADR 0614 +// §4), and the direct writes to the store, accepted from the mesh until every node's have of every build module runs a +// build that publishes through the seats. Each wakes at distribution's start, on each `delivered` and on the node-engine +// seat's `manifest-reads-changed` (ADR 0618 §4), never on a timer; each reads every node, away ones included, through +// node_haves and `manifest-reads`, never a watch of another module's state (ADR 0596). +const ( + OldNameWindow = "old-name" + DirectWritesWindow = "direct-writes" + // WindowFindingAfter is how long a node holds a window open before it is distribution's finding naming the node. + WindowFindingAfter = 7 * 24 * time.Hour + // The events a window's close says on distribution's own family distribution.events (7 days, 16 MiB, drop + // oldest), never on a seat's `.events`. + OldNameClosedEvent = "old-name-closed" + DirectWritesClosedEvent = "direct-writes-closed" + // NewNamesCapability is the reader capability a node-engine states when it reads the new seat names (ADR 0614 §4). + NewNamesCapability = "seat:" + ImageSeat +) + +// Window is one window's record, kept in distribution's state windows: what registry_old_name and +// registry_direct_writes answer. +type Window struct { + Name string `json:"name"` + Open bool `json:"open"` + Holding []string `json:"holding,omitempty"` + Since map[string]time.Time `json:"since,omitempty"` + Dropped []DroppedNode `json:"dropped,omitempty"` + Closed time.Time `json:"closed,omitzero"` + // LastHolding are the nodes that held it last, said when it closes. + LastHolding []string `json:"last_holding,omitempty"` +} + +// DroppedNode is one retired node dropped from a window by hand: a hand act. +type DroppedNode struct { + Node string `json:"node"` + By string `json:"by"` + Why string `json:"why"` + At time.Time `json:"at"` +} + +// Closed is the body of a window's closing event. +type Closed struct { + Window string `json:"window"` + Seat string `json:"seat,omitempty"` + At time.Time `json:"at"` + LastHolding []string `json:"last_holding"` + Dropped []DroppedNode `json:"dropped,omitempty"` +} + +// NodeRead is what distribution read of one node: the reader capabilities its node-engine states through +// `manifest-reads` (Serves false while it does not serve the verb: it states nothing), whether its declaration names an +// artifact, and its build modules' haves: whether each runs a build that publishes through the seats. +type NodeRead struct { + Node string `json:"node"` + Serves bool `json:"serves"` + Stated []string `json:"stated,omitempty"` + NamesArtifact bool `json:"names_artifact"` + BuildModules map[string]bool `json:"build_modules,omitempty"` +} + +// HoldingOldName are the nodes holding the old name open: a node whose declaration names an artifact and whose +// node-engine does not state the new names (one not serving manifest-reads states nothing). +func HoldingOldName(nodes []NodeRead, dropped []DroppedNode) []string { + var out []string + for _, n := range nodes { + if isDropped(dropped, n.Node) || !n.NamesArtifact { + continue + } + if !n.Serves || !slices.Contains(n.Stated, NewNamesCapability) { + out = append(out, n.Node) + } + } + sort.Strings(out) + return out +} + +// HoldingDirectWrites are the nodes one of whose build modules' haves runs a build that does not yet publish through +// the seats. +func HoldingDirectWrites(nodes []NodeRead, dropped []DroppedNode) []string { + var out []string + for _, n := range nodes { + if isDropped(dropped, n.Node) { + continue + } + for _, publishes := range n.BuildModules { + if !publishes { + out = append(out, n.Node) + break + } + } + } + sort.Strings(out) + return out +} + +func isDropped(dropped []DroppedNode, node string) bool { + return slices.ContainsFunc(dropped, func(d DroppedNode) bool { return d.Node == node }) +} + +// Step moves the window to the nodes holding it at now. It answers the next record, the closing event when it closed in +// this step (said once), and the nodes holding it past WindowFindingAfter (the finding; cleared when it closes). A +// window once closed stays closed: a node that holds it later is no reason to serve the old name again. +func (w Window) Step(holding []string, now time.Time) (next Window, closed *Closed, overdue []string) { + next = Window{Name: w.Name, Dropped: w.Dropped, Closed: w.Closed, LastHolding: w.LastHolding} + if !w.Open && !w.Closed.IsZero() { + return next, nil, nil + } + if len(holding) == 0 { + next.Closed, next.LastHolding = now, w.Holding + if next.LastHolding == nil { + next.LastHolding = []string{} + } + return next, &Closed{Window: w.Name, At: now, LastHolding: next.LastHolding, Dropped: w.Dropped}, nil + } + next.Open, next.Holding, next.Since = true, holding, map[string]time.Time{} + for _, n := range holding { + since, had := w.Since[n] + if !had { + since = now + } + next.Since[n] = since + if now.Sub(since) >= WindowFindingAfter { + overdue = append(overdue, n) + } + } + return next, nil, overdue +} + +// Drop drops a retired node from the window by hand, a hand act with who and why. +func (w Window) Drop(node, by, why string, now time.Time) (Window, error) { + if node == "" || why == "" { + return w, fmt.Errorf("drop: a node and why are required (a hand act)") + } + if isDropped(w.Dropped, node) { + return w, fmt.Errorf("drop: %s is dropped from %s already", node, w.Name) + } + w.Dropped = append(w.Dropped, DroppedNode{Node: node, By: by, Why: why, At: now}) + return w, nil +}