From 1b1019972d9f94d25b61f742377c5fbb5d3a39c6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:44:39 +0100 Subject: [PATCH] Hold the seat artifact-store in artifact-store-tools: the keep set from the store's listing, the wants and each module's five newest builds, the sweep at a build's end once the lease names no controller with each deletion recorded, and S25's image-missing (issue 1955, ADR 0571 clause 2) --- .../cmd/artifact-store-tools/door.go | 116 +++ .../cmd/artifact-store-tools/holder.go | 810 ++++++++++++++++++ .../holder_integration_test.go | 245 ++++++ .../cmd/artifact-store-tools/holder_test.go | 286 +++++++ .../cmd/artifact-store-tools/main.go | 28 +- .../cmd/artifact-store-tools/seat.go | 134 +++ .../cmd/artifact-store-tools/store_test.go | 29 +- .../cmd/artifact-store-tools/tools.go | 11 +- .../artifact-store-tools/files/store-address | 1 + modules/artifact-store-tools/go.mod | 21 +- modules/artifact-store-tools/go.sum | 29 +- .../internal/registry/blob.go | 57 ++ .../internal/registry/blob_test.go | 44 + .../internal/registry/hold.go | 345 ++++++++ .../internal/registry/hold_test.go | 268 ++++++ .../internal/registry/platforms_test.go | 176 ++++ .../internal/registry/store.go | 325 +++++++ .../internal/registry/store_test.go | 134 +++ modules/artifact-store-tools/module.json | 55 +- 19 files changed, 3086 insertions(+), 28 deletions(-) create mode 100644 modules/artifact-store-tools/cmd/artifact-store-tools/door.go create mode 100644 modules/artifact-store-tools/cmd/artifact-store-tools/holder.go create mode 100644 modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go create mode 100644 modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go create mode 100644 modules/artifact-store-tools/cmd/artifact-store-tools/seat.go create mode 100644 modules/artifact-store-tools/files/store-address create mode 100644 modules/artifact-store-tools/internal/registry/blob.go create mode 100644 modules/artifact-store-tools/internal/registry/blob_test.go create mode 100644 modules/artifact-store-tools/internal/registry/hold.go create mode 100644 modules/artifact-store-tools/internal/registry/hold_test.go create mode 100644 modules/artifact-store-tools/internal/registry/platforms_test.go create mode 100644 modules/artifact-store-tools/internal/registry/store.go create mode 100644 modules/artifact-store-tools/internal/registry/store_test.go diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go new file mode 100644 index 00000000..bd730cfe --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go @@ -0,0 +1,116 @@ +package main + +// The artifact store's door and the bus, as the holder reaches them through this module's runtime. + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "artifactstoretools/internal/registry" +) + +// door is registry.Store at the store's address as this machine reaches it. +type door struct{ s registry.Store } + +func (d *door) Address() string { return d.s.Address } +func (d *door) Hold(ctx context.Context, r string) (bool, error) { return d.s.Hold(ctx, r) } +func (d *door) Holds(ctx context.Context, r string) error { return d.s.Holds(ctx, r) } +func (d *door) IsIndex(ctx context.Context, r string) (bool, error) { + return d.s.IsIndex(ctx, r) +} +func (d *door) LetGo(ctx context.Context, r string) error { return d.s.LetGo(ctx, r) } +func (d *door) SpareKept(ctx context.Context, keptImages, touching []string) error { + spare, err := d.s.SpareKeptIndexes(ctx, keptImages, touching) + if err == nil { + d.s.Spare = spare + } + return err +} + +// DoorAt reads the store's address (host:port) from the file the manifest fills from the seat mesh-artifact-store's +// reach, each time, so a store that moved is reached where it is now. +func DoorAt(file string) func() (Door, error) { + return func() (Door, error) { + if file == "" { + return nil, fmt.Errorf("this bundle was given no file naming the artifact store's address") + } + raw, err := os.ReadFile(file) + if err != nil { + return nil, fmt.Errorf("the artifact store's address: %w", err) + } + address := strings.TrimSpace(string(raw)) + if address == "" { + return nil, fmt.Errorf("the artifact store's address is empty in %s", file) + } + return &door{s: registry.Store{Address: address}}, nil + } +} + +// runtimeBus is the holder's bus through the runtime: its reads, its findings and its seat's event. +type runtimeBus struct{} + +func (runtimeBus) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +func (runtimeBus) Get(state, key string) ([]byte, bool, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} +func (runtimeBus) History(state, key string) ([]stdio.StateRevision, error) { + return stdio.State(state).History(key) +} +func (runtimeBus) Put(state, key string, value any) error { + _, err := stdio.State(state).Put(key, value) + return err +} +func (runtimeBus) Delete(state, key string) error { return stdio.State(state).Delete(key) } +func (runtimeBus) Publish(subject string, body any) error { + return stdio.SeatPublish(subject, body, "") +} + +// follow watches the builder's done keys and the wants: a build's end sweeps (once the lease names no controller) and +// observes again; a want changed observes again. Event-driven: nothing reads them on a timer. +func follow(h *Holder) { + <-stdio.Served() + watch := func(state string, handle func(stdio.StateChange) error) { + for { + if err := stdio.State(state).Watch("", handle); err == nil { + return + } else { + h.logf("watching %s: %v; asked again in 10s", state, err) + } + time.Sleep(10 * time.Second) + } + } + observe := func() { + if err := h.Observe(context.Background()); err != nil { + h.logf("image-missing could not be observed: %v", err) + } + } + watch(BuildsRead, func(c stdio.StateChange) error { + if c.Current || c.Op != stdio.OpPut { + return nil + } + var b struct { + State string `json:"state"` + } + if json.Unmarshal(c.Value, &b) == nil && b.State == "done" { + go func() { h.AfterBuild(c.Key); observe() }() + } + return nil + }) + watch(WantsRead, func(c stdio.StateChange) error { + if !c.Current { + go observe() + } + return nil + }) + observe() +} diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go new file mode 100644 index 00000000..bfdb1ecf --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go @@ -0,0 +1,810 @@ +package main + +// **The seat artifact-store** (novox/hq ADR 0571 clause 2, issue 1955): this module holds the artifact store's sweep, out +// of the controller. Its keep set is the store's own listing joined with every want's fingerprint (the assigner's +// `wants`) and each module's five newest builds (the builder's done keys, each build's end read from the key's history; +// ADR 0526 §1): a digest a want names is kept however old; a build beyond the five newest that no want names is +// eligible; a manifest in the store that no build names is unrecorded and never removed (ADR 0189 §10, ADR 0251 §6). +// +// At each build's end (a done key written) it sweeps as the controller's take-in did: every kept archive held first +// (ADR 0251 §8), then what is eligible let go of, oldest first, within the same bound per sweep, an index left for a +// `collect` a person confirms (ADR 0251 §7), each deletion recorded in the hand-act log (seat mesh-hand-acts) and said +// as artifacts-gone. **Nothing is let go until the controller's lease names a holder that is no controller**: until +// then the controller's take-in sweeps, and two sweeps never run at once. It also observes S25, image-missing: a want's +// image the store does not hold, a finding in its own state `findings` the condition keeper reads. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/health" + + "artifactstoretools/internal/registry" +) + +// What the holder reads and keeps. +const ( + WantsRead = delivery.AssignerModule + ".wants" + BuildsRead = delivery.BuilderModule + ".builds" + HavesRead = "node-engine.have" + LeaseRead = buscompose.ControllerLease + // FindingsState is the holder's own findings (health.FindingsState): image-missing. + FindingsState = health.FindingsState + // Module is this module, the observer its findings name. + Module = "artifact-store-tools" + // GoneSubject is where the seat says artifacts-gone. + GoneSubject = "mesh.seat." + catalogue.ArtifactStoreSeat + ".event." + delivery.ArtifactsGoneEvent +) + +// The sweep's bounds, the controller's (cmd/mesh-controller/collect.go mostPerSweep, sweepBudget). +const ( + KeptBuilds = 5 + MostPerSweep = 500 + MostAtAll = 5000 + SweepBudget = 45 * time.Second + ImageMissing = "image-missing" +) + +// Bus is what the holder reads and writes on the bus: its read states (wants, builds, haves, the lease), its own +// findings, and its seat's event. The runtime in main; a test's in memory. +type Bus interface { + Keys(state string) ([]string, error) + Get(state, key string) ([]byte, bool, error) + History(state, key string) ([]stdio.StateRevision, error) + Put(state, key string, value any) error + Delete(state, key string) error + Publish(subject string, body any) error +} + +// Door is the artifact store's own door: registry.Store, behind an interface so a test needs no store. +type Door interface { + Address() string + Hold(ctx context.Context, reference string) (bool, error) + Holds(ctx context.Context, reference string) error + IsIndex(ctx context.Context, reference string) (bool, error) + LetGo(ctx context.Context, reference string) error + // SpareKept sets the platform manifests a kept index names aside before an index is let go of (ADR 0257). + SpareKept(ctx context.Context, keptImages, touching []string) error +} + +// Holder is the seat's holder. +type Holder struct { + Bus Bus + Door func() (Door, error) + // Listed is the store's own listing now (Store.View); nil reads none, and then nothing is said gone or collected. + Listed func(ctx context.Context) (*View, error) + Acts handacts.Recorder + Now func() time.Time + Logf func(string, ...any) + + sweep sync.Mutex // one sweep at a time + mu sync.Mutex + ended map[string]time.Time // a done key's end, read once from its history +} + +// Build is one done build of a module, as its key holds it, with when it ended. +type Build struct { + Module, FP, Key string + At time.Time + Artifacts []delivery.BuiltArtifact +} + +// LeaseAway says the controller's lease names a holder that is no controller: then the seat sweeps. +func (h *Holder) LeaseAway() bool { + raw, found, err := h.Bus.Get(LeaseRead, buscompose.ControllerLeaseKey) + away, _ := buscompose.LeaseHandedAway(raw, found, err) + return away +} + +// Builds is every done build the builder's keys hold, each with its end. +func (h *Holder) Builds() ([]Build, error) { + keys, err := h.Bus.Keys(BuildsRead) + if err != nil { + return nil, fmt.Errorf("the builder's builds: %w", err) + } + var out []Build + for _, k := range keys { + module, fp, err := delivery.ParseBuilderBuildKey(k) + if err != nil { + continue + } + raw, found, err := h.Bus.Get(BuildsRead, k) + if err != nil { + return nil, fmt.Errorf("the builder's build %s: %w", k, err) + } + if !found { + continue + } + b, err := delivery.DecodeBuilderBuild(raw) + if err != nil || b.State != delivery.BuildDone { + continue + } + at, err := h.endOf(k) + if err != nil { + return nil, err + } + out = append(out, Build{Module: module, FP: fp, Key: k, At: at, Artifacts: b.Artifacts}) + } + return out, nil +} + +// endOf is when a done key was last written done, read once from its history: a done key is not written again. +func (h *Holder) endOf(key string) (time.Time, error) { + h.mu.Lock() + at, known := h.ended[key] + h.mu.Unlock() + if known { + return at, nil + } + revs, err := h.Bus.History(BuildsRead, key) + if err != nil { + return time.Time{}, fmt.Errorf("the history of the builder's build %s: %w", key, err) + } + for _, r := range revs { + if b, err := delivery.DecodeBuilderBuild(r.Value); err == nil && b.State == delivery.BuildDone && r.At.After(at) { + at = r.At + } + } + h.mu.Lock() + if h.ended == nil { + h.ended = map[string]time.Time{} + } + h.ended[key] = at + h.mu.Unlock() + return at, nil +} + +// Want is one node's want for a module. +type Want struct{ Node, Module, FP string } + +// Wants is every want that names a fingerprint. +func (h *Holder) Wants() ([]Want, error) { + keys, err := h.Bus.Keys(WantsRead) + if err != nil { + return nil, fmt.Errorf("the wants: %w", err) + } + var out []Want + for _, k := range keys { + node, module, ok := wantKey(k) + if !ok { + continue + } + raw, found, err := h.Bus.Get(WantsRead, k) + if err != nil { + return nil, fmt.Errorf("the want %s: %w", k, err) + } + var w delivery.Want + if !found || json.Unmarshal(raw, &w) != nil || w.Removed || w.FP == "" { + continue + } + out = append(out, Want{Node: node, Module: module, FP: w.FP}) + } + return out, nil +} + +// wantKey reads `want..`; the node's own key and a hold are not a module's want. +func wantKey(key string) (node, module string, ok bool) { + parts := strings.Split(key, ".") + if len(parts) != 3 || parts[0] != "want" || parts[2] == "node" { + return "", "", false + } + return parts[1], parts[2], true +} + +// Ref is one artifact the mesh made, as the keep set judges it. +type Ref struct { + Reference string + Kind string // image or archive + Builds []Build + Why []string + Kept bool + // At is its newest build's end: the oldest go first. + At time.Time +} + +// KeepSet judges every artifact the done builds name in the store: kept, and why, or eligible. Packages and +// references into no store of the mesh's are no business of the store's. +func KeepSet(builds []Build, wants []Want) map[string]*Ref { + wanted := map[string][]string{} + for _, w := range wants { + wanted[w.Module+" "+w.FP] = append(wanted[w.Module+" "+w.FP], w.Node) + } + byModule := map[string][]Build{} + for _, b := range builds { + byModule[b.Module] = append(byModule[b.Module], b) + } + keeps := map[string][]string{} // build key → why + for module, bs := range byModule { + sort.Slice(bs, func(i, j int) bool { + if !bs[i].At.Equal(bs[j].At) { + return bs[i].At.After(bs[j].At) + } + return bs[i].Key > bs[j].Key + }) + for i, b := range bs { + if i < KeptBuilds { + keeps[b.Key] = append(keeps[b.Key], fmt.Sprintf("one of the %d newest builds of %s", KeptBuilds, module)) + } + if nodes := wanted[module+" "+b.FP]; len(nodes) > 0 { + sort.Strings(nodes) + keeps[b.Key] = append(keeps[b.Key], "a want names it on "+strings.Join(nodes, ", ")) + } + } + } + out := map[string]*Ref{} + for _, b := range builds { + for _, a := range b.Artifacts { + reference := catalogue.Recorded(a.Reference) + if _, ours := catalogue.InArtifactStore(reference); !ours || a.Kind == "package" { + continue + } + r := out[reference] + if r == nil { + _, _, kind := splitReference(reference) + r = &Ref{Reference: reference, Kind: kind} + out[reference] = r + } + r.Builds = append(r.Builds, b) + if b.At.After(r.At) { + r.At = b.At + } + if why := keeps[b.Key]; len(why) > 0 { + r.Kept = true + for _, w := range why { + if !contains(r.Why, w) { + r.Why = append(r.Why, w) + } + } + } + } + } + return out +} + +func contains(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} + +// inStore says whether the listing holds a reference: an image's manifest, an archive's blob. Unknown without a listing. +func inStore(v *View, reference string) (held, known bool) { + if v == nil || v.L == nil { + return false, false + } + repo, digest, kind := splitReference(reference) + switch kind { + case "image": + r := v.L.Repos[repo] + return r != nil && r.Revisions[digest], true + case "archive": + _, has := v.L.Blobs[digest] + return has, true + } + return false, false +} + +// judged is the keep set now, with the listing it was read against. +type judged struct { + refs map[string]*Ref + view *View + wants []Want + built []Build +} + +func (h *Holder) judge(ctx context.Context) (*judged, error) { + builds, err := h.Builds() + if err != nil { + return nil, err + } + wants, err := h.Wants() + if err != nil { + return nil, err + } + j := &judged{refs: KeepSet(builds, wants), wants: wants, built: builds} + if h.Listed != nil { + if j.view, err = h.Listed(ctx); err != nil { + h.logf("the store's listing could not be read, so nothing is said gone: %v", err) + j.view = nil + } + } + return j, nil +} + +// Artifacts is the seat's `artifacts`, in the controller's answer's shape (Records): every reference the mesh made that +// lives in the store, kept with why or eligible, and with collected the ones the store no longer holds. +func (h *Holder) Artifacts(repository string) (*Records, error) { + j, err := h.judge(context.Background()) + if err != nil { + return nil, err + } + r := j.records(repository, true) + if d, err := h.Door(); err == nil { + r.Store = d.Address() + } + return r, nil +} + +func (j *judged) records(repository string, collected bool) *Records { + r := &Records{KeptBuilds: KeptBuilds, Counts: map[string]int{"kept": 0, "eligible": 0, "collected": 0}, CollectedAsked: collected} + for _, ref := range j.sorted() { + state := "eligible" + if ref.Kept { + state = "kept" + } + if held, known := inStore(j.view, ref.Reference); known && !held { + state = "collected" + if ref.Kept { + state = "kept" // a kept one gone is said missing by collect, never let go of + } + } + r.Counts[state]++ + if (repository != "" && !strings.HasPrefix(strings.TrimPrefix(ref.Reference, catalogue.ArtifactStoreScheme), repository)) || + (state == "collected" && !collected) { + continue + } + rec := Record{Reference: ref.Reference, Kind: ref.Kind, State: state, Why: ref.Why} + fill(&rec) + r.References = append(r.References, rec) + } + if j.view == nil { + r.Note = "the store's listing could not be read, so a reference it no longer holds is not said collected" + } + return r +} + +// sorted is the references oldest first, then by reference. +func (j *judged) sorted() []*Ref { + out := make([]*Ref, 0, len(j.refs)) + for _, r := range j.refs { + out = append(out, r) + } + sort.Slice(out, func(a, b int) bool { + if !out[a].At.Equal(out[b].At) { + return out[a].At.Before(out[b].At) + } + return out[a].Reference < out[b].Reference + }) + return out +} + +// Collect is the seat's `collect`, in the controller's answer's shape (CollectAnswer): a dry run unless confirm; with +// confirm and why, a sweep that also lets indexes go with their platform manifests. +func (h *Holder) Collect(why string, confirm bool, most int) (*CollectAnswer, error) { + if confirm && strings.TrimSpace(why) == "" { + return nil, errors.New("a confirmed collect needs why: nothing was let go of") + } + if most <= 0 { + most = MostPerSweep + } + if most > MostAtAll { + return nil, fmt.Errorf("collect lets go of at most %d at once, not %d", MostAtAll, most) + } + if confirm && !h.LeaseAway() { + return nil, errors.New("the controller holds its lease, and its take-in sweeps the store until it hands it away: " + + "this seat lets nothing go before then (novox/hq ADR 0571 clause 2); ask the controller's collect") + } + return h.Sweep(context.Background(), Sweeping{Confirmed: confirm, DryRun: !confirm, Most: most, Why: why, By: "collect"}) +} + +// Sweeping is one sweep's asks. +type Sweeping struct { + // Confirmed is a collect a person confirmed: an index goes with its platform manifests. + Confirmed bool + // DryRun lets nothing go and holds nothing: what would go is listed. + DryRun bool + Most int + Why string + // By names what asked: collect, or the build whose end it follows. + By string +} + +// Sweep holds every kept archive, then lets go of what is eligible, oldest first, within the bounds, recording each +// deletion and saying artifacts-gone. +func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) { + h.sweep.Lock() + defer h.sweep.Unlock() + j, err := h.judge(ctx) + if err != nil { + return nil, err + } + a := &CollectAnswer{DryRun: s.DryRun, WouldLetGo: []string{}, LetGo: []string{}} + var keptArchives, keptImages, eligible []string + for _, r := range j.sorted() { + switch { + case r.Kept && r.Kind == "archive": + keptArchives = append(keptArchives, r.Reference) + case r.Kept: + keptImages = append(keptImages, r.Reference) + default: + if held, known := inStore(j.view, r.Reference); known && !held { + continue // collected already + } + eligible = append(eligible, r.Reference) + } + } + a.KeptArchives, a.Eligible = len(keptArchives), len(eligible) + if s.DryRun { + for i, r := range eligible { + if i >= s.Most { + a.Left = len(eligible) - i + break + } + a.WouldLetGo = append(a.WouldLetGo, r) + } + return a, nil + } + door, err := h.Door() + if err != nil { + return nil, err + } + a.Store = door.Address() + within, stop := context.WithTimeout(ctx, SweepBudget) + defer stop() + // Hold before letting go (ADR 0251 §8): a kept archive that cannot be held stops the sweep before any delete. + var missing []string + for _, r := range keptArchives { + wrote, err := door.Hold(within, r) + switch { + case err == nil: + if wrote { + a.HoldersWritten++ + } + case errors.Is(err, registry.Gone): + missing = append(missing, r) + case errors.Is(err, registry.ErrNotOurs): + default: + a.Stopped = fmt.Sprintf("not every kept archive could be held, so nothing was let go: holding %s: %v", r, err) + a.Left = len(eligible) + return a, nil + } + } + a.Held, a.Missing = len(keptArchives)-len(missing), len(missing) + h.sayGone(j, missing, "a kept archive the store does not hold") + if s.Confirmed { + if err := door.SpareKept(within, keptImages, eligible); err != nil { + a.Stopped = fmt.Sprintf("which platform manifests a kept index names could not be read, so nothing was let go: %v", err) + a.Left = len(eligible) + return a, nil + } + } + for i, r := range eligible { + if i >= s.Most || within.Err() != nil { + a.Left = len(eligible) - i + if i >= s.Most { + a.Stopped = fmt.Sprintf("asked about %d, the most one sweep asks", s.Most) + } else { + a.Stopped = fmt.Sprintf("ran out of its %s", SweepBudget) + } + break + } + if !s.Confirmed { + // An index waits for a confirmed collect (ADR 0251 §7, ADR 0257 §4). + index, err := door.IsIndex(within, r) + if err != nil && !errors.Is(err, registry.Gone) && !errors.Is(err, registry.ErrNotOurs) { + a.Stopped = fmt.Sprintf("the store could not say whether %s is an index, so nothing more was asked: %v", r, err) + a.Left = len(eligible) - i + break + } + if index { + a.Skipped++ + continue + } + } + err := door.LetGo(within, r) + switch { + case err == nil || errors.Is(err, registry.Gone): + case errors.Is(err, registry.ErrNotOurs): + a.Skipped++ + continue + default: + a.Stopped = fmt.Sprintf("the store refused to let go of %s, so nothing more was asked: %v", r, err) + a.Left = len(eligible) - i + } + if a.Stopped != "" { + break + } + // Recorded in the hand-act log, each one (ADR 0565 §10; the operator's "option A"). + why := s.Why + if why == "" { + why = "eligible: no want names it and it is not among its module's " + fmt.Sprint(KeptBuilds) + " newest builds" + } + if _, err := h.Acts.Record(handacts.Request{What: "let go of " + r + " (" + s.By + ")", Why: why, + Cause: "artifact-store-sweep", Verb: "collect"}); err != nil { + h.logf("%s was let go of and could not be recorded in the hand-act log: %v", r, err) + } + a.LetGo = append(a.LetGo, r) + } + h.sayGone(j, a.LetGo, "let go of") + return a, nil +} + +// sayGone says artifacts-gone for each build that named one of refs: the builder builds a key again whose artifact is gone. +func (h *Holder) sayGone(j *judged, refs []string, how string) { + byKey := map[string]*delivery.ArtifactsGone{} + var keys []string + for _, r := range refs { + ref := j.refs[r] + if ref == nil { + continue + } + for _, b := range ref.Builds { + g := byKey[b.Key] + if g == nil { + g = &delivery.ArtifactsGone{Module: b.Module, Key: b.Key, Fingerprint: b.FP, At: h.now().UTC()} + byKey[b.Key] = g + keys = append(keys, b.Key) + } + g.Gone = append(g.Gone, r) + } + } + sort.Strings(keys) + for _, k := range keys { + raw, err := delivery.EncodeArtifactsGone(*byKey[k]) + if err == nil { + err = h.Bus.Publish(GoneSubject, json.RawMessage(raw)) + } + if err != nil { + h.logf("%s: %s %s, and artifacts-gone could not be said: %v", k, how, strings.Join(byKey[k].Gone, ", "), err) + } + } +} + +// AfterBuild is the sweep at a build's end: only once the lease names no controller. +func (h *Holder) AfterBuild(key string) { + if !h.LeaseAway() { + return + } + a, err := h.Sweep(context.Background(), Sweeping{Most: MostPerSweep, By: "after the build " + key}) + if err != nil { + h.logf("the sweep after %s: %v", key, err) + return + } + if len(a.LetGo) > 0 || a.Stopped != "" || a.Missing > 0 { + h.logf("the sweep after %s let go of %d (%d left, %d missing): %s", key, len(a.LetGo), a.Left, a.Missing, a.Stopped) + } +} + +// Held is the seat's `held`: each reference, held, gone or unknown with why. Holds and writes nothing. +func (h *Holder) Held(refs []string) (delivery.HeldAnswer, error) { + if len(refs) == 0 { + return delivery.HeldAnswer{}, errors.New("held judges the references given, and none was") + } + if len(refs) > 64 { + return delivery.HeldAnswer{}, fmt.Errorf("held judges at most 64 references at once, and %d were given", len(refs)) + } + for _, r := range refs { + if _, ours := catalogue.InArtifactStore(r); !ours { + return delivery.HeldAnswer{}, fmt.Errorf("%s is not a reference into the mesh's artifact store (artifact-store://...): nothing was asked", r) + } + } + door, err := h.Door() + if err != nil { + return delivery.HeldAnswer{}, err + } + asking, stop := context.WithTimeout(context.Background(), 20*time.Second) + defer stop() + a := delivery.HeldAnswer{Store: door.Address(), Held: []string{}, Gone: []string{}} + for _, r := range refs { + switch err := door.Holds(asking, r); { + case err == nil: + a.Held = append(a.Held, r) + case errors.Is(err, registry.Gone): + a.Gone = append(a.Gone, r) + default: + if a.Unknown == nil { + a.Unknown = map[string]string{} + } + a.Unknown[r] = err.Error() + } + } + return a, nil +} + +// Image is one image a machine's wants and haves name, as docker reads the answer. +type Image struct { + Image string `json:"image"` + Resources []string `json:"resources"` + In []string `json:"in"` +} + +// Images is the seat's `images`, the controller's answer's shape: every image one machine's wants (in: declared) and +// haves (in: sent) name, by the builds their fingerprints name, each with the module naming it. +type Images struct { + Node string `json:"node"` + Images []Image `json:"images"` + SentKnown bool `json:"sent_known"` +} + +// ImagesOf answers `images` for node. +func (h *Holder) ImagesOf(node string, address string) (Images, error) { + out := Images{Node: node, Images: []Image{}} + wants, err := h.Wants() + if err != nil { + return out, err + } + byImage := map[string]*Image{} + add := func(module, fp, in string) error { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return nil + } + raw, found, err := h.Bus.Get(BuildsRead, key) + if err != nil || !found { + return err + } + b, err := delivery.DecodeBuilderBuild(raw) + if err != nil { + return nil + } + for _, a := range b.Artifacts { + if a.Kind != "image" && a.Kind != "upstream" || a.Reference == "" { + continue + } + image := catalogue.Routed(catalogue.Recorded(a.Reference), address) + i := byImage[image] + if i == nil { + i = &Image{Image: image} + byImage[image] = i + } + if !contains(i.Resources, module) { + i.Resources = append(i.Resources, module) + } + if !contains(i.In, in) { + i.In = append(i.In, in) + } + } + return nil + } + for _, w := range wants { + if w.Node == node { + if err := add(w.Module, w.FP, "declared"); err != nil { + return out, err + } + } + } + keys, err := h.Bus.Keys(HavesRead) + if err != nil { + return out, fmt.Errorf("the haves: %w", err) + } + for _, k := range keys { + n, module, isNode, err := delivery.ParseHaveKey(k) + if err != nil || isNode || n != node { + continue + } + raw, found, err := h.Bus.Get(HavesRead, k) + if err != nil { + return out, err + } + var have delivery.Have + if !found || json.Unmarshal(raw, &have) != nil || have.FP == "" { + continue + } + out.SentKnown = true + if err := add(module, have.FP, "sent"); err != nil { + return out, err + } + } + for _, i := range byImage { + sort.Strings(i.Resources) + sort.Strings(i.In) + out.Images = append(out.Images, *i) + } + sort.Slice(out.Images, func(a, b int) bool { return out.Images[a].Image < out.Images[b].Image }) + return out, nil +} + +// Observe writes S25's findings (image-missing): each want whose build names an image the store does not hold, on the +// want's machine, and deletes each finding that no longer holds. Nothing is said without a listing. +func (h *Holder) Observe(ctx context.Context) error { + j, err := h.judge(ctx) + if err != nil { + return err + } + if j.view == nil { + return nil + } + builds := map[string]Build{} + for _, b := range j.built { + builds[b.Module+" "+b.FP] = b + } + now := map[string]health.Finding{} + for _, w := range j.wants { + b, ok := builds[w.Module+" "+w.FP] + if !ok { + continue + } + for _, a := range b.Artifacts { + reference := catalogue.Recorded(a.Reference) + if a.Kind != "image" { + continue + } + if held, known := inStore(j.view, reference); !known || held { + continue + } + f := imageMissing(w.Node, w.Module, reference, h.now()) + now[f.Key] = f + break + } + } + keys, err := h.Bus.Keys(FindingsState) + if err != nil { + return fmt.Errorf("the findings: %w", err) + } + for _, k := range keys { + if _, still := now[k]; !still { + if err := h.Bus.Delete(FindingsState, k); err != nil { + return err + } + } + } + for k, f := range now { + raw, found, err := h.Bus.Get(FindingsState, k) + if err != nil { + return err + } + if found { + if was, err := health.DecodeFinding(raw); err == nil && was.Summary == f.Summary { + continue // written only when what it says changes + } + } + if _, err := health.EncodeFinding(f); err != nil { + return err + } + if err := h.Bus.Put(FindingsState, k, f); err != nil { + return err + } + } + return nil +} + +// imageMissing is S25's finding, in the controller's words (cmd/mesh-controller/image_missing.go watchMissingImages). +func imageMissing(node, module, image string, at time.Time) health.Finding { + id := module + "." + node + return health.Finding{V: health.FindingVersion, Key: health.FindingKey(health.ScopeModule, id, ImageMissing), Kind: ImageMissing, + Subject: health.Subject{Scope: health.ScopeModule, ID: id, Machine: node}, + Severity: health.Warning, Resolver: health.ResolverOperator, + Summary: fmt.Sprintf("%s is left out of every send to %s: its declaration names an image that the artifact store does "+ + "not hold (the evidence names it); the running container is untouched", module, node), + Headline: capital(module) + " is not updated on " + node, + Explanation: fmt.Sprintf("The new version of %s names a container image that the mesh's store does not hold, so it "+ + "was not sent to %s. What runs there is untouched, and the rest of %s is sent as usual.", module, node, node), + Needs: fmt.Sprintf("build %s again; the details name the image.", module), + Resolved: "Resolved: the store holds the image of " + module + " again", + Evidence: []health.Evidence{{At: at.UTC(), Said: "image " + image + " is not in the store"}}, + Source: "S25", Observer: Module, At: at.UTC()} +} + +func capital(s string) string { + if s == "" { + return s + } + return strings.ToUpper(s[:1]) + s[1:] +} + +func (h *Holder) now() time.Time { + if h.Now != nil { + return h.Now() + } + return time.Now() +} + +func (h *Holder) logf(format string, a ...any) { + if h.Logf != nil { + h.Logf(format, a...) + } +} diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go new file mode 100644 index 00000000..c71a6a58 --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go @@ -0,0 +1,245 @@ +//go:build integration + +package main + +// The seat artifact-store's holder whole against a real bus (novox/hq ADR 0571 clause 2, issue 1955): the builder's done +// keys, the wants and the controller's lease in a real nats-server's key-value buckets, a build's end heard by a watch +// of the builds as the runtime hands it over, each deletion recorded by a request to the seat mesh-hand-acts on the bus, +// artifacts-gone said on the seat's subject, and S25's finding written to the holder's own bucket. Only the store's +// door is in memory: its HTTP is the copied registry package's, tested there. + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// realBus is the holder's Bus over a real nats-server: each read state a bucket named as the runtime names it. +type realBus struct { + ctx context.Context + nc *nats.Conn + js jetstream.JetStream +} + +func bucketOf(state string) string { return strings.Replace(state, ".", "_", 1) } + +func (b realBus) kv(state string) (jetstream.KeyValue, error) { + return b.js.KeyValue(b.ctx, bucketOf(state)) +} + +func (b realBus) Keys(state string) ([]string, error) { + kv, err := b.kv(state) + if err != nil { + return nil, err + } + keys, err := kv.Keys(b.ctx) + if err == jetstream.ErrNoKeysFound { + return nil, nil + } + return keys, err +} + +func (b realBus) Get(state, key string) ([]byte, bool, error) { + kv, err := b.kv(state) + if err != nil { + return nil, false, err + } + e, err := kv.Get(b.ctx, key) + if err == jetstream.ErrKeyNotFound { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return e.Value(), true, nil +} + +func (b realBus) History(state, key string) ([]stdio.StateRevision, error) { + kv, err := b.kv(state) + if err != nil { + return nil, err + } + h, err := kv.History(b.ctx, key) + if err != nil { + return nil, err + } + var out []stdio.StateRevision + for _, e := range h { + out = append(out, stdio.StateRevision{Key: key, Value: e.Value(), Revision: e.Revision(), At: e.Created()}) + } + return out, nil +} + +func (b realBus) Put(state, key string, value any) error { + kv, err := b.kv(state) + if err != nil { + return err + } + raw, _ := json.Marshal(value) + _, err = kv.Put(b.ctx, key, raw) + return err +} + +func (b realBus) Delete(state, key string) error { + kv, err := b.kv(state) + if err != nil { + return err + } + return kv.Delete(b.ctx, key) +} + +func (b realBus) Publish(subject string, body any) error { + raw, _ := json.Marshal(body) + return b.nc.Publish(subject, raw) +} + +func TestTheSeatSweepsAtABuildsEndOnARealBusOnceTheLeaseIsAway(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + for _, s := range []string{BuildsRead, WantsRead, HavesRead, LeaseRead, "artifact-store-tools." + FindingsState} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucketOf(s), History: 5}); err != nil { + t.Fatal(err) + } + } + bus := realBus{ctx: ctx, nc: nc, js: js} + + // The mesh as aMesh lays it, copied onto the real buckets: letta's seven builds, a want, the store's listing. + mem, memb, door, _ := aMesh(t, &buscompose.LeaseHolder{Instance: "c1", Host: "novox", Epoch: 3}) + for _, s := range []string{BuildsRead, WantsRead, LeaseRead} { + keys, _ := memb.Keys(s) + for _, k := range keys[:min(len(keys), 6)] { // the seventh build ends later, on the bus + raw, _, _ := memb.Get(s, k) + if err := bus.Put(s, k, json.RawMessage(raw)); err != nil { + t.Fatal(err) + } + } + } + time.Sleep(20 * time.Millisecond) // each build's end is its key's time on the bus: the seventh the newest + + // The seat mesh-hand-acts, answering every act it records; artifacts-gone heard on the seat's subject. + var mu sync.Mutex + var acts, gone []string + if _, err := nc.Subscribe("mesh.seat."+handacts.Seat+".tool.hand-act", func(m *nats.Msg) { + mu.Lock() + acts = append(acts, string(m.Data)) + mu.Unlock() + _ = m.Respond([]byte(`{"ok":true}`)) + }); err != nil { + t.Fatal(err) + } + if _, err := nc.Subscribe(GoneSubject, func(m *nats.Msg) { + mu.Lock() + gone = append(gone, string(m.Data)) + mu.Unlock() + }); err != nil { + t.Fatal(err) + } + h := &Holder{Bus: findingsAt{realBus: bus}, Door: mem.Door, Listed: mem.Listed, Now: time.Now, Logf: t.Logf} + h.Acts = handacts.Recorder{LeaseAway: h.LeaseAway, Ask: func(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + m, err := nc.Request("mesh.seat."+strings.TrimPrefix(strings.Replace(key, ".", ".tool.", 1), "seat:"), raw, 5*time.Second) + if err != nil { + return nil, err + } + return m.Data, nil + }} + + // A build's end heard from the builds bucket, as follow hands it over. + kv, _ := bus.kv(BuildsRead) + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + t.Fatal(err) + } + ended := func() { + keys, _ := memb.Keys(BuildsRead) + raw, _, _ := memb.Get(BuildsRead, keys[6]) + if err := bus.Put(BuildsRead, keys[6], json.RawMessage(raw)); err != nil { + t.Fatal(err) + } + select { + case e := <-w.Updates(): + h.AfterBuild(e.Key()) + case <-ctx.Done(): + t.Fatal("the build's end was not heard") + } + } + + // While the lease names the controller, a build's end lets nothing go. + ended() + if len(door.asked) != 0 { + t.Fatalf("before the lease write the store was asked %v", door.asked) + } + // The lease handed away: the next build's end holds the kept archives and lets go of letta's 2nd, which the 7th moved + // past the five newest and no want names (the want keeps the 1st), recording each deletion. + if err := bus.Put(LeaseRead, buscompose.ControllerLeaseKey, buscompose.LeaseHolder{HandedTo: "mesh-grants", Why: "the lease write"}); err != nil { + t.Fatal(err) + } + ended() + deadline := time.Now().Add(5 * time.Second) + for { + mu.Lock() + n := len(gone) + mu.Unlock() + if n > 0 || time.Now().After(deadline) { + break + } + time.Sleep(20 * time.Millisecond) + } + mu.Lock() + defer mu.Unlock() + if len(acts) != 2 || !strings.Contains(acts[0], archiveRef("letta", 2)) || !strings.Contains(acts[1], imageRef("letta", 2)) { + t.Fatalf("the hand acts recorded are %v; the store was asked %v", acts, door.asked) + } + if len(gone) != 1 || !strings.Contains(gone[0], `"fingerprint":"`+fp(2)+`"`) { + t.Fatalf("artifacts-gone said %v", gone) + } + + // S25 through the real findings bucket: letta's wanted image missing from the store is a finding there. + view, _ := mem.Listed(ctx) + repo, digest, _ := splitReference(imageRef("letta", 1)) + delete(view.L.Repos[repo].Revisions, digest) + if err := h.Observe(ctx); err != nil { + t.Fatal(err) + } + if keys, _ := bus.Keys("artifact-store-tools." + FindingsState); len(keys) != 1 || keys[0] != "module.letta.ace.image-missing" { + t.Fatalf("the findings are %v", keys) + } +} + +// findingsAt reads and writes the holder's own findings in its bucket, artifact-store-tools_findings. +type findingsAt struct{ realBus } + +func (f findingsAt) own(state string) string { + if state == FindingsState { + return Module + "." + state + } + return state +} +func (f findingsAt) Keys(state string) ([]string, error) { return f.realBus.Keys(f.own(state)) } +func (f findingsAt) Get(state, key string) ([]byte, bool, error) { + return f.realBus.Get(f.own(state), key) +} +func (f findingsAt) Put(state, key string, value any) error { + return f.realBus.Put(f.own(state), key, value) +} +func (f findingsAt) Delete(state, key string) error { return f.realBus.Delete(f.own(state), key) } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go new file mode 100644 index 00000000..94d51300 --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go @@ -0,0 +1,286 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/health" + + "artifactstoretools/internal/registry" +) + +// memBus is the holder's bus in memory: states by name, each key's history, and what was published. +type memBus struct { + mu sync.Mutex + states map[string]map[string][]byte + history map[string][]stdio.StateRevision + published []string +} + +func (b *memBus) Keys(state string) ([]string, error) { + b.mu.Lock() + defer b.mu.Unlock() + var out []string + for k := range b.states[state] { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func (b *memBus) Get(state, key string) ([]byte, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + v, ok := b.states[state][key] + return v, ok, nil +} + +func (b *memBus) History(state, key string) ([]stdio.StateRevision, error) { + return b.history[state+" "+key], nil +} + +func (b *memBus) Put(state, key string, value any) error { + raw, _ := json.Marshal(value) + b.mu.Lock() + defer b.mu.Unlock() + if b.states[state] == nil { + b.states[state] = map[string][]byte{} + } + b.states[state][key] = raw + return nil +} + +func (b *memBus) Delete(state, key string) error { + b.mu.Lock() + defer b.mu.Unlock() + delete(b.states[state], key) + return nil +} + +func (b *memBus) Publish(subject string, body any) error { + raw, _ := json.Marshal(body) + b.mu.Lock() + defer b.mu.Unlock() + b.published = append(b.published, subject+" "+string(raw)) + return nil +} + +// memDoor is the store's door in memory: what it holds, which references are indexes, and what was asked. +type memDoor struct { + holds map[string]bool + indexes map[string]bool + asked []string +} + +func (d *memDoor) Address() string { return "10.0.0.5:5000" } +func (d *memDoor) Hold(_ context.Context, r string) (bool, error) { + d.asked = append(d.asked, "hold "+r) + if !d.holds[r] { + return false, registry.Gone + } + return false, nil +} +func (d *memDoor) Holds(_ context.Context, r string) error { + if !d.holds[r] { + return registry.Gone + } + return nil +} +func (d *memDoor) IsIndex(_ context.Context, r string) (bool, error) { return d.indexes[r], nil } +func (d *memDoor) LetGo(_ context.Context, r string) error { + d.asked = append(d.asked, "let go "+r) + delete(d.holds, r) + return nil +} +func (d *memDoor) SpareKept(context.Context, []string, []string) error { + d.asked = append(d.asked, "spare") + return nil +} + +func digestOf(ref string) string { _, d, _ := splitReference(ref); return d } + +func fp(n int) string { return "src2:" + strings.Repeat(fmt.Sprintf("%x", n%16), 64) } + +func imageRef(module string, n int) string { + return "artifact-store://" + module + "/image@sha256:" + strings.Repeat(fmt.Sprintf("%x", n%16), 64) +} + +func archiveRef(module string, n int) string { + return "artifact-store://" + module + "/files/blobs/sha256:" + strings.Repeat(fmt.Sprintf("%x", (n+8)%16), 64) +} + +// A mesh with seven done builds of letta, the n-th ending n hours after a day began, each with an image and an archive; +// a want names the first on ace; the store lists them all and one image no build names; the lease as given. +func aMesh(t *testing.T, lease *buscompose.LeaseHolder) (*Holder, *memBus, *memDoor, *[]string) { + t.Helper() + day := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + b := &memBus{states: map[string]map[string][]byte{}, history: map[string][]stdio.StateRevision{}} + door := &memDoor{holds: map[string]bool{}, indexes: map[string]bool{}} + listing := &Listing{Blobs: map[string]int64{}, Repos: map[string]*Repo{}} + for n := 1; n <= 7; n++ { + key, _ := delivery.BuilderBuildKey("letta", fp(n)) + raw, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: strings.Repeat("d", 64), Artifacts: []delivery.BuiltArtifact{ + {Name: "image", Kind: "image", Digest: digestOf(imageRef("letta", n)), Reference: imageRef("letta", n)}, + {Name: "files", Kind: "archive", Digest: digestOf(archiveRef("letta", n)), Reference: archiveRef("letta", n)}, + {Name: "pkg", Kind: "package", Digest: "sha256:" + strings.Repeat("f", 64), Reference: "letta@1.0." + fmt.Sprint(n)}}}) + if err != nil { + t.Fatal(err) + } + _ = b.Put(BuildsRead, key, json.RawMessage(raw)) + b.history[BuildsRead+" "+key] = []stdio.StateRevision{{Key: key, Value: raw, At: day.Add(time.Duration(n) * time.Hour)}} + for _, r := range []string{imageRef("letta", n), archiveRef("letta", n)} { + door.holds[r] = true + repo, digest, kind := splitReference(r) + if kind == "archive" { + listing.Blobs[digest] = 10 + } else { + listing.repo(repo).Revisions[digest] = true + } + } + } + listing.repo("letta/image").Revisions["sha256:"+strings.Repeat("e", 64)] = true // no build names it + want, _ := json.Marshal(delivery.Want{V: delivery.Version, FP: fp(1), Self: true}) + _ = b.Put(WantsRead, "want.ace.letta", json.RawMessage(want)) + if lease != nil { + _ = b.Put(LeaseRead, buscompose.ControllerLeaseKey, lease) + } + var acts []string + h := &Holder{Bus: b, Door: func() (Door, error) { return door, nil }, + Listed: func(context.Context) (*View, error) { return &View{L: listing}, nil }, + Now: func() time.Time { return day.Add(48 * time.Hour) }, + Acts: handacts.Recorder{Ask: func(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + acts = append(acts, key+" "+string(raw)) + return json.RawMessage(`{}`), nil + }}} + h.Acts.LeaseAway = h.LeaseAway + return h, b, door, &acts +} + +// ADR 0571 clause 2: the keep set from the store's listing, the wants and the five newest builds; nothing let go of while +// the lease names a controller; once it names another holder, a build's end lets go of the eligible within the bound, +// leaves an index for a confirmed collect, records each deletion and says artifacts-gone. +func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsOnlyOnceTheLeaseIsAway(t *testing.T) { + controller := &buscompose.LeaseHolder{Instance: "c1", Host: "novox", Epoch: 3} + handed := &buscompose.LeaseHolder{HandedTo: "mesh-grants", HandedBy: "person.jochen", Why: "the lease write"} + + h, _, _, _ := aMesh(t, controller) + r, err := h.Artifacts("") + if err != nil { + t.Fatal(err) + } + states := map[string]string{} + for _, x := range r.References { + states[x.Reference] = x.State + } + // letta's 3rd to 7th are the five newest; its 1st a want names on ace, however old; its 2nd is eligible; the package + // and the manifest no build names are no reference of the seat's. + for n, want := range map[int]string{1: "kept", 2: "eligible", 3: "kept", 7: "kept"} { + if states[imageRef("letta", n)] != want || states[archiveRef("letta", n)] != want { + t.Errorf("letta's build %d is %s and %s, not %s", n, states[imageRef("letta", n)], states[archiveRef("letta", n)], want) + } + } + if len(r.References) != 14 || r.Counts["kept"] != 12 || r.Counts["eligible"] != 2 || r.Store != "10.0.0.5:5000" { + t.Errorf("the references are %d, counted %v, at %q", len(r.References), r.Counts, r.Store) + } + + for _, c := range []struct { + name string + lease *buscompose.LeaseHolder + index bool + most int + letGo []string + left int + skipped int + }{ + {name: "the lease names a controller", lease: controller}, + {name: "no lease key", lease: nil}, + {name: "handed away", lease: handed, most: MostPerSweep, letGo: []string{archiveRef("letta", 2), imageRef("letta", 2)}}, + {name: "an index waits", lease: handed, index: true, most: MostPerSweep, letGo: []string{archiveRef("letta", 2)}, skipped: 1}, + {name: "bounded", lease: handed, most: 1, letGo: []string{archiveRef("letta", 2)}, left: 1}, + } { + t.Run(c.name, func(t *testing.T) { + h, b, door, acts := aMesh(t, c.lease) + door.indexes[imageRef("letta", 2)] = c.index + if c.most == 0 { + h.AfterBuild("letta.x") + if _, err := h.Collect("tidy", true, 0); err == nil { + t.Error("a confirmed collect was let through while the controller sweeps") + } + if len(door.asked) != 0 || len(*acts) != 0 || len(b.published) != 0 { + t.Fatalf("before the lease write the store was asked %v, acts %v, said %v", door.asked, *acts, b.published) + } + return + } + a, err := h.Sweep(context.Background(), Sweeping{Most: c.most, By: "after the build letta.x"}) + if err != nil { + t.Fatal(err) + } + if fmt.Sprint(a.LetGo) != fmt.Sprint(c.letGo) || a.Left != c.left || a.Skipped != c.skipped || a.Held != 6 || a.Missing != 0 { + t.Fatalf("let go of %v (left %d, skipped %d, held %d, missing %d), want %v", a.LetGo, a.Left, a.Skipped, a.Held, a.Missing, c.letGo) + } + // Every kept archive held before the first deletion. + first := -1 + for i, x := range door.asked { + if strings.HasPrefix(x, "let go ") && first < 0 { + first = i + } + } + if first != 6 { + t.Errorf("the store was asked %v: the six kept archives are held first", door.asked) + } + if len(*acts) != len(c.letGo) || !strings.HasPrefix((*acts)[0], "seat:mesh-hand-acts.hand-act ") || + !strings.Contains((*acts)[0], c.letGo[0]) { + t.Errorf("the hand acts recorded are %v", *acts) + } + if len(b.published) != 1 || !strings.HasPrefix(b.published[0], GoneSubject+" ") || + !strings.Contains(b.published[0], `"fingerprint":"`+fp(2)+`"`) { + t.Errorf("said %v", b.published) + } + }) + } +} + +// S25 (image-missing): a want whose build names an image the store does not hold is a finding in the seat's own state, +// written once and deleted once the store holds it again; images names what a machine's wants and haves name. +func TestAWantedImageTheStoreLacksIsAFindingAndImagesNamesWantsAndHaves(t *testing.T) { + h, b, _, _ := aMesh(t, nil) + view, _ := h.Listed(context.Background()) + repo, digest, _ := splitReference(imageRef("letta", 1)) + delete(view.L.Repos[repo].Revisions, digest) + h.Listed = func(context.Context) (*View, error) { return view, nil } + if err := h.Observe(context.Background()); err != nil { + t.Fatal(err) + } + key := health.FindingKey(health.ScopeModule, "letta.ace", ImageMissing) + raw, found, _ := b.Get(FindingsState, key) + f, err := health.DecodeFinding(raw) + if !found || err != nil || f.Subject.Machine != "ace" || f.Source != "S25" || !strings.Contains(f.Evidence[0].Said, imageRef("letta", 1)) { + t.Fatalf("the finding is %+v (%v, %v)", f, found, err) + } + view.L.Repos[repo].Revisions[digest] = true + if err := h.Observe(context.Background()); err != nil { + t.Fatal(err) + } + if _, found, _ := b.Get(FindingsState, key); found { + t.Fatal("the finding stayed once the store held the image again") + } + + have, _ := json.Marshal(delivery.Have{V: delivery.Version, FP: fp(7), Phase: "running"}) + _ = b.Put(HavesRead, "ace.letta", json.RawMessage(have)) + images, err := h.ImagesOf("ace", "10.0.0.5:5000") + if err != nil || len(images.Images) != 2 || !images.SentKnown || images.Images[0].Image != "10.0.0.5:5000/letta/image@sha256:"+strings.Repeat("1", 64) || + fmt.Sprint(images.Images[0].In) != "[declared]" || fmt.Sprint(images.Images[1].In) != "[sent]" { + t.Fatalf("the images of ace are %+v: %v", images, err) + } +} diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go index f833fecf..5deb0bb7 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go @@ -1,30 +1,38 @@ -// The artifact-store-tools module's Go bundle (novox/hq ADR 0251 §1–3, to-be 51): a process the node's -// runtime launches and speaks MCP over stdio to. It says what the artifact store holds — its -// repositories, its size, and what the controller's records say of each manifest — read from the -// store's own files through the store's own container, and writing nothing. A collection is asked of -// the controller, which decides and records what it lets go of (ADR 0189). +// The artifact-store-tools module's Go bundle (novox/hq ADR 0251 §1–3, ADR 0571 clause 2, to-be 51): a process the +// node's runtime launches and speaks MCP over stdio to. It says what the artifact store holds — its repositories, its +// size, and what the mesh keeps of each manifest — read from the store's own files through the store's own container, +// and it holds the seat artifact-store: the keep set, the sweep at each build's end once the controller's lease names +// no controller, and S25's image-missing. // -// **A module beside the store, not the store's own.** A module that provides the artifact store cannot -// also build an artifact: building publishes to the store, so the store would be needed to create -// itself, and the controller refuses that manifest. So these tools are a second module, assigned to the -// machine that holds the store. stdout is the protocol; what this bundle says, it says on stderr. +// **A module beside the store, not the store's own.** A module that provides the artifact store cannot also build an +// artifact: building publishes to the store, so the store would be needed to create itself. So these tools are a +// second module, assigned to the machine that holds the store. stdout is the protocol; what this bundle says, it says +// on stderr. package main import ( "fmt" "os" + "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" ) +func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } + func main() { s := &Store{ Container: os.Getenv("MESH_ARTIFACT_STORE_CONTAINER"), Run: ExecRunner, UID: os.Getuid(), } + h := &Holder{Bus: runtimeBus{}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View, + Now: time.Now, Logf: logf} + h.Acts = handacts.Recorder{Ask: stdio.Ask, LeaseAway: h.LeaseAway} + go follow(h) // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): artifact-store-tools. - if err := stdio.Serve("", Tools(s, Controller{Ask: stdio.Ask}, &Catalogue{Ask: stdio.Ask})); err != nil { + if err := stdio.Serve("", append(Tools(s, h, &Catalogue{Ask: stdio.Ask}), SeatTools(h, s)...)); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/seat.go b/modules/artifact-store-tools/cmd/artifact-store-tools/seat.go new file mode 100644 index 00000000..6e8fe4cc --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/seat.go @@ -0,0 +1,134 @@ +package main + +// The seat artifact-store's verbs (novox/hq ADR 0571 clause 2; catalogue.ArtifactStoreSeat), each a tool named +// `artifact-store.` as the runtime serves a seat's verb, answering as the controller's verb answered: {output, +// ok, answer}. + +import ( + "context" + "fmt" + "sort" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// verbAnswer is the controller's verbs' answer shape, which the builder and docker read (answerOf). +type verbAnswer struct { + Output string `json:"output"` + OK bool `json:"ok"` + Answer any `json:"answer"` +} + +// SeatTools are the seat's verbs over h; s reads the store's listing for mirrors. +func SeatTools(h *Holder, s *Store) []stdio.Tool { + seat := catalogue.ArtifactStoreSeat + "." + return []stdio.Tool{ + {Name: seat + "artifacts", Description: "Every artifact the mesh made that the store holds: kept and why, eligible, " + + "or with collected those it no longer holds.", + Run: func(args map[string]any) (any, error) { + r, err := h.Artifacts(text(args, "repository")) + if err != nil { + return nil, err + } + if text(args, "collected") != "true" { + refs := r.References[:0:0] + for _, x := range r.References { + if x.State != "collected" { + refs = append(refs, x) + } + } + r.References = refs + } + return verbAnswer{Output: fmt.Sprintf("%d kept, %d eligible, %d collected\n", r.Counts["kept"], + r.Counts["eligible"], r.Counts["collected"]), OK: true, Answer: r}, nil + }}, + {Name: seat + "held", Description: "Whether the store holds each reference given: held, gone or unknown with why.", + Run: func(args map[string]any) (any, error) { + a, err := h.Held(listOf(args["references"])) + if err != nil { + return nil, err + } + return verbAnswer{Output: fmt.Sprintf("the artifact store at %s holds %d of the %d reference(s) asked\n", + a.Store, len(a.Held), len(a.Held)+len(a.Gone)+len(a.Unknown)), OK: true, Answer: a}, nil + }}, + {Name: seat + "collect", Description: "The sweep now: a dry run without confirm; with confirm and why, every kept " + + "archive held, then each eligible artifact let go of, each recorded in the hand-act log.", + Run: func(args map[string]any) (any, error) { + most, err := count(args, "most", MostPerSweep, 1, MostAtAll) + if err != nil { + return nil, err + } + a, err := h.Collect(text(args, "why"), text(args, "confirm") == "true", most) + if err != nil { + return nil, err + } + return verbAnswer{Output: fmt.Sprintf("let go of %d, would let go of %d, %d left\n", len(a.LetGo), + len(a.WouldLetGo), a.Left), OK: true, Answer: a}, nil + }}, + {Name: seat + "mirrors", Description: "The bases the mesh copied into its store, as the store lists them.", + Run: func(args map[string]any) (any, error) { + if text(args, "record") != "" { + return nil, fmt.Errorf("mirrors records nothing on this seat: a copy is kept or let go of by the builds " + + "the builder's log says stood on it (novox/hq ADR 0571 clause 1), and until that log names them every " + + "copy is unrecorded and never removed") + } + v, err := s.View(context.Background()) + if err != nil { + return nil, err + } + var copies []string + for _, name := range v.L.RepoNames() { + if !mirrorRepository.MatchString(name) { + continue + } + for d := range v.L.Repos[name].Revisions { + copies = append(copies, catalogue.ArtifactStoreScheme+name+"@"+d) + } + } + sort.Strings(copies) + return verbAnswer{Output: fmt.Sprintf("%d copies, each unrecorded and never removed\n", len(copies)), OK: true, + Answer: map[string]any{"unrecorded": nonNil(copies)}}, nil + }}, + {Name: seat + "images", Description: "Every container image one machine's wants and haves name.", + Run: func(args map[string]any) (any, error) { + node := text(args, "node") + if node == "" { + return nil, fmt.Errorf("images names the machine") + } + address := "" + if d, err := h.Door(); err == nil { + address = d.Address() + } + a, err := h.ImagesOf(node, address) + if err != nil { + return nil, err + } + return verbAnswer{Output: fmt.Sprintf("%s: %d images\n", node, len(a.Images)), OK: true, Answer: a}, nil + }}, + } +} + +// listOf is a list argument: a list, or one text separated by commas. +func listOf(v any) []string { + var out []string + add := func(s string) { + for _, p := range strings.Split(s, ",") { + if p = strings.TrimSpace(p); p != "" && !contains(out, p) { + out = append(out, p) + } + } + } + switch x := v.(type) { + case string: + add(x) + case []any: + for _, e := range x { + if s, ok := e.(string); ok { + add(s) + } + } + } + return out +} diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go index 72b46379..fc6413e4 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go @@ -1,6 +1,8 @@ package main import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "context" "encoding/json" "fmt" @@ -411,6 +413,11 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { var m struct { Tools []string `json:"tools"` Invokes []string `json:"invokes"` + Reads []string `json:"reads"` + Claims []struct { + Name string `json:"name"` + Serves []string `json:"serves"` + } `json:"claims"` } if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) @@ -428,10 +435,30 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { if !reflect.DeepEqual(served, listed) { t.Fatalf("served %v, manifest %v", served, listed) } - want := []string{"seat:mesh-controller.artifacts", "seat:mesh-controller.collect", CatalogueModules, CatalogueModule} + // ADR 0571 clause 2: it holds the seat artifact-store and serves its verbs, and calls no verb of the controller's; + // the lease is the one controller state it reads (clause 10's exemption). + want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule} if !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes %v", m.Invokes) } + var verbs []string + for _, tl := range SeatTools(&Holder{}, &Store{}) { + verbs = append(verbs, strings.TrimPrefix(tl.Name, catalogue.ArtifactStoreSeat+".")) + } + seat, _ := catalogue.SeatNamed(catalogue.ArtifactStoreSeat) + var contract []string + for _, v := range seat.Serves { + contract = append(contract, v.Name) + } + if len(m.Claims) != 1 || m.Claims[0].Name != catalogue.ArtifactStoreSeat || !reflect.DeepEqual(m.Claims[0].Serves, verbs) || + !reflect.DeepEqual(verbs, contract) { + t.Errorf("claims %+v, serves %v, the seat's verbs %v", m.Claims, verbs, contract) + } + for _, r := range m.Reads { + if strings.HasPrefix(r, "mesh-controller.") && r != LeaseRead { + t.Errorf("reads the controller's %s", r) + } + } } func TestASocketRefusalIsAskedAgainThroughSudo(t *testing.T) { diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/tools.go b/modules/artifact-store-tools/cmd/artifact-store-tools/tools.go index 65d0350e..a6656750 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/tools.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/tools.go @@ -11,9 +11,16 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" ) +// Recorded is what says which references the mesh keeps and lets go of them: the seat's holder (Holder, novox/hq ADR +// 0571 clause 2), or before it the controller (Controller), each in the controller's answer's shape. +type Recorded interface { + Artifacts(repository string) (*Records, error) + Collect(why string, confirm bool, most int) (*CollectAnswer, error) +} + // Tools are the artifact store's tools (novox/hq ADR 0251 §1). The first three only read; artifact_store_collect -// only reads too, unless it is a real run, and then it asks the controller, which decides and deletes. -func Tools(s *Store, c Controller, cat *Catalogue) []stdio.Tool { +// only reads too, unless it is a real run, and then it asks the seat's sweep, which decides and deletes. +func Tools(s *Store, c Recorded, cat *Catalogue) []stdio.Tool { ctx := context.Background() repoArg := map[string]any{"type": "string", "description": "one repository, as the store names it (/)"} return []stdio.Tool{ diff --git a/modules/artifact-store-tools/files/store-address b/modules/artifact-store-tools/files/store-address new file mode 100644 index 00000000..1bcadab2 --- /dev/null +++ b/modules/artifact-store-tools/files/store-address @@ -0,0 +1 @@ +${seat:mesh-artifact-store:reach} diff --git a/modules/artifact-store-tools/go.mod b/modules/artifact-store-tools/go.mod index 74ccd152..d2d7c1fa 100644 --- a/modules/artifact-store-tools/go.mod +++ b/modules/artifact-store-tools/go.mod @@ -1,5 +1,22 @@ module artifactstoretools -go 1.22 +go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.7 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266 + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/modules/artifact-store-tools/go.sum b/modules/artifact-store-tools/go.sum index b474419e..ad48fccc 100644 --- a/modules/artifact-store-tools/go.sum +++ b/modules/artifact-store-tools/go.sum @@ -1,2 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w= -git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266 h1:32lx0DCHm+7Ig6t8vXbfwPbk4YgbHQt1sRzmRlezFWw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223859-85f5fd0b2266/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/artifact-store-tools/internal/registry/blob.go b/modules/artifact-store-tools/internal/registry/blob.go new file mode 100644 index 00000000..bedbfa38 --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/blob.go @@ -0,0 +1,57 @@ +package registry + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net/http" +) + +// blobBound is the most GetBlob reads: a templates archive is a few kilobytes; anything near this is no archive a +// take-in reads text from. +const blobBound = 64 << 20 + +// GetBlob is the archive a recorded reference names (`artifact-store:///blobs/sha256:`), read whole +// and checked against the digest it names: bytes that hash to another digest are refused, never read (R4e: the +// controller reads a templates archive by the reference of a builder's done key). Gone when the store does not hold +// it; ErrNotOurs for a reference outside the store or one that is not an archive. +func (s Store) GetBlob(ctx context.Context, reference string) ([]byte, error) { + repository, digest, archive, err := s.archive(reference) + if err != nil { + return nil, err + } + if !archive { + return nil, fmt.Errorf("%w: %s is not an archive", ErrNotOurs, reference) + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url(repository, "blobs", digest), nil) + if err != nil { + return nil, err + } + response, err := s.client().Do(request) + if err != nil { + return nil, fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err) + } + defer response.Body.Close() + switch response.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return nil, Gone + default: + return nil, fmt.Errorf("the artifact store answered %s for %s/blobs/%s", response.Status, repository, digest) + } + body, err := io.ReadAll(io.LimitReader(response.Body, blobBound+1)) + if err != nil { + return nil, fmt.Errorf("reading %s/blobs/%s: %w", repository, digest, err) + } + if len(body) > blobBound { + return nil, fmt.Errorf("%s/blobs/%s is larger than %d bytes, which no archive read here is", repository, digest, blobBound) + } + sum := sha256.Sum256(body) + if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest { + return nil, fmt.Errorf("the artifact store answered %s/blobs/%s with bytes of the digest %s: refused", repository, + digest, got) + } + return body, nil +} diff --git a/modules/artifact-store-tools/internal/registry/blob_test.go b/modules/artifact-store-tools/internal/registry/blob_test.go new file mode 100644 index 00000000..13487cc7 --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/blob_test.go @@ -0,0 +1,44 @@ +package registry + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// A take-in reads a templates archive by its reference, and only bytes of the digest the reference names. +func TestGetBlobReadsOnlyTheBytesOfItsDigest(t *testing.T) { + good := []byte("the archive") + sum := sha256.Sum256(good) + digest := "sha256:" + hex.EncodeToString(sum[:]) + liar := "sha256:" + strings.Repeat("0", 64) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/x/files/blobs/"+digest), strings.HasSuffix(r.URL.Path, "/x/files/blobs/"+liar): + _, _ = w.Write(good) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + s := Store{Address: strings.TrimPrefix(server.URL, "http://")} + ctx := context.Background() + body, err := s.GetBlob(ctx, "artifact-store://x/files/blobs/"+digest) + if err != nil || string(body) != string(good) { + t.Fatalf("the archive: %q %v", body, err) + } + if _, err := s.GetBlob(ctx, "artifact-store://x/files/blobs/"+liar); err == nil || !strings.Contains(err.Error(), "refused") { + t.Fatalf("bytes of another digest are read: %v", err) + } + if _, err := s.GetBlob(ctx, "artifact-store://x/other/blobs/"+digest); !errors.Is(err, Gone) { + t.Fatalf("a blob the store does not hold: %v", err) + } + if _, err := s.GetBlob(ctx, "registry.example/x@"+digest); !errors.Is(err, ErrNotOurs) { + t.Fatalf("a reference outside the store: %v", err) + } +} diff --git a/modules/artifact-store-tools/internal/registry/hold.go b/modules/artifact-store-tools/internal/registry/hold.go new file mode 100644 index 00000000..391e9c89 --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/hold.go @@ -0,0 +1,345 @@ +package registry + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Every archive the store keeps is held by a manifest (novox/hq issue 253, ADR 0189). +// +// **The store's collector marks only from manifests.** The mesh's store is a stock registry, and +// its nightly `registry garbage-collect` walks every manifest in every repository, marks the blobs +// those manifests name, and deletes every blob it did not mark. An image is a manifest, so what +// the mesh keeps of an image survives. An archive was not: the builder put it in the store as a +// bare blob — upload, then `PUT ?digest=` — and nothing in the store names it. To the collector a +// bare blob is unreferenced, so the first real collection would have deleted every archive the +// mesh holds, kept or not, and every machine pinning a bundle would have found it gone. The +// collector runs `--dry-run` until this is true. +// +// **So each archive gets a holder**: the smallest OCI image manifest that names it — the empty +// config, one layer, nothing else — put in the archive's own repository, by digest, untagged. The +// collector marks it and so keeps the archive; the sweep lets go of an archive by deleting its +// holder first, which is what lets the bytes go at the next collection. +// +// **Nothing a machine reads changes.** The recorded reference stays +// `artifact-store:////blobs/sha256:…`, and machines fetch the blob exactly as +// before. The holder is the store's bookkeeping, not a second way to reach anything. +// +// **Deterministic, so it never needs recording.** The holder is composed from the archive's digest +// and size alone, in a fixed field order with no timestamps or annotations, so the sweep can +// compute which manifest holds any archive from the reference it already has plus one HEAD for the +// size. No schema change, no second record that could disagree with the store. + +const ( + // mediaManifest is the type a holder is put and asked for as. + mediaManifest = "application/vnd.oci.image.manifest.v1+json" + // mediaEmpty is the OCI empty descriptor's type: a config that says nothing, for a manifest + // whose only purpose is to name its layer. + mediaEmpty = "application/vnd.oci.empty.v1+json" + // emptyDigest is the digest of `{}`, the empty config's content, fixed by the OCI spec. + emptyDigest = "sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a" + // mediaArchive is the layer type an archive is held as. Every archive the builder publishes is + // `pack`'s gzipped tar, so this is the true type and not a placeholder — and it is a constant, + // not read from anywhere, because the holder must be recomputable from the reference alone. + mediaArchive = "application/vnd.oci.image.layer.v1.tar+gzip" +) + +// emptyConfig is the content emptyDigest names. +var emptyConfig = []byte("{}") + +// manifestAccept is what a manifest is asked for as. A registry answers a manifest HEAD only in a +// type the caller named, and answers 404 to a bare one for a manifest it holds perfectly well +// (measured 2026-09-28; internal/builder/registry.go says how that was found). +var manifestAccept = []string{ + mediaManifest, + "application/vnd.docker.distribution.manifest.v2+json", +} + +type descriptor struct { + MediaType string `json:"mediaType"` + Digest string `json:"digest"` + Size int64 `json:"size"` +} + +type holderManifest struct { + SchemaVersion int `json:"schemaVersion"` + MediaType string `json:"mediaType"` + Config descriptor `json:"config"` + Layers []descriptor `json:"layers"` +} + +// Holder is the manifest that holds an archive in the store, and its digest. +// +// A pure function of the archive's digest and size: the same two in give the same bytes out, +// always, because `encoding/json` writes a struct's fields in their declared order and there is +// nothing here that varies by when or where it was composed. +func Holder(digest string, size int64) (body []byte, holder string) { + body, err := json.Marshal(holderManifest{ + SchemaVersion: 2, + MediaType: mediaManifest, + Config: descriptor{MediaType: mediaEmpty, Digest: emptyDigest, Size: int64(len(emptyConfig))}, + Layers: []descriptor{{MediaType: mediaArchive, Digest: digest, Size: size}}, + }) + if err != nil { + // Marshalling a struct of strings and integers cannot fail. + panic(err) + } + sum := sha256.Sum256(body) + return body, "sha256:" + hex.EncodeToString(sum[:]) +} + +// Hold makes sure the store holds this archive by a manifest, and says whether it had to write one. +// +// Takes a reference as the mesh records it. An image is its own manifest and needs no holder, so +// it answers false and nothing is asked. Idempotent: a holder already there is left alone, which +// is what lets the sweep run it over every kept archive on every build and so backfill the bare +// blobs published before holders existed (novox/hq issue 253). +// +// Gone when the store does not hold the archive at all: there is nothing to hold, and that is a +// fact the caller reports rather than one this invents a remedy for. +func (s Store) Hold(ctx context.Context, reference string) (bool, error) { + repository, digest, archive, err := s.archive(reference) + if err != nil || !archive { + return false, err + } + size, err := s.blobSize(ctx, repository, digest) + if err != nil { + return false, err + } + return s.HoldBlob(ctx, repository, digest, size) +} + +// Held is whether the store holds this archive by its manifest. Asks and changes nothing — the +// question an operator needs answered with "none unheld" before the collector is let loose. +// +// An image answers true: it is its own manifest. An archive the store does not have answers Gone. +func (s Store) Held(ctx context.Context, reference string) (bool, error) { + repository, digest, archive, err := s.archive(reference) + if err != nil { + return false, err + } + if !archive { + return true, nil + } + size, err := s.blobSize(ctx, repository, digest) + if err != nil { + return false, err + } + _, holder := Holder(digest, size) + return s.has(ctx, s.url(repository, "manifests", holder), manifestAccept...) +} + +// HoldBlob puts the holder for a blob of this digest and size into its repository, unless it is +// there already. Answers whether it wrote one. +// +// The builder calls this with the size it has just uploaded; the sweep, through Hold, with the size +// the store reports. Both arrive at the same holder, which is the point of composing it. +func (s Store) HoldBlob(ctx context.Context, repository, digest string, size int64) (bool, error) { + if s.Address == "" { + return false, fmt.Errorf("this mesh has no artifact store on its network to hold %s/%s in", repository, digest) + } + body, holder := Holder(digest, size) + there, err := s.has(ctx, s.url(repository, "manifests", holder), manifestAccept...) + if err != nil { + return false, err + } + if there { + return false, nil + } + + // The config must be in the repository before a manifest naming it is accepted: a registry + // refuses a manifest whose blobs it cannot find there, which is the property that makes a + // holder mean something. + if err := s.putBlob(ctx, repository, emptyDigest, emptyConfig); err != nil { + return false, err + } + + // **By digest, never by tag.** A tag would be one more name to move and one more thing the + // collector's `--delete-untagged` would read as meaningful; the mesh names nothing by tag that + // it pins by digest, and an untagged manifest is kept by plain collection. + request, err := http.NewRequestWithContext(ctx, http.MethodPut, + s.url(repository, "manifests", holder), bytes.NewReader(body)) + if err != nil { + return false, err + } + request.Header.Set("Content-Type", mediaManifest) + response, err := s.client().Do(request) + if err != nil { + return false, err + } + defer response.Body.Close() + if response.StatusCode != http.StatusCreated { + said, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + return false, fmt.Errorf("the artifact store refused to hold %s/%s: %s %s", + repository, digest, response.Status, strings.TrimSpace(string(said))) + } + return true, nil +} + +// letGoOfHolder deletes the manifest holding an archive, before the archive's own link goes. +// +// **Holder first.** Deleting the blob link alone leaves a manifest still naming the blob, and the +// collector would keep its bytes for ever on the strength of it — the sweep would record the +// archive collected while the disk said otherwise. Deleting the holder first and failing before +// the link goes leaves an unheld archive that the next sweep still offers, which is safe. +// +// A store that no longer has the blob answers Gone: without its size the holder cannot be named, +// and without the blob there is nothing left for a holder to keep. A store that never had a holder +// for it — an archive published before holders, never backfilled — answers 404 to the delete, and +// that is the outcome wanted. +func (s Store) letGoOfHolder(ctx context.Context, repository, digest string) error { + size, err := s.blobSize(ctx, repository, digest) + if err != nil { + return err + } + _, holder := Holder(digest, size) + err = s.remove(ctx, s.url(repository, "manifests", holder), repository+"/manifests/"+holder) + if err == Gone { + return nil + } + return err +} + +// archive reads a recorded reference into its repository and digest, and whether it is an archive +// at all. Refuses as ErrNotOurs anything the mesh did not put in its own store. +func (s Store) archive(reference string) (repository, digest string, archive bool, err error) { + path, kept := catalogue.InArtifactStore(reference) + if !kept { + return "", "", false, fmt.Errorf("%w: %s", ErrNotOurs, reference) + } + if s.Address == "" { + return "", "", false, fmt.Errorf("this mesh has no artifact store on its network to ask about %s", reference) + } + repository, kind, digest, err := split(path) + if err != nil { + return "", "", false, err + } + return repository, digest, kind == "blobs", nil +} + +// blobSize is how large the store says a blob is; Gone when it does not have it. +func (s Store) blobSize(ctx context.Context, repository, digest string) (int64, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodHead, s.url(repository, "blobs", digest), nil) + if err != nil { + return 0, err + } + response, err := s.client().Do(request) + if err != nil { + return 0, fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err) + } + defer response.Body.Close() + switch response.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return 0, Gone + default: + return 0, fmt.Errorf("the artifact store answered %s for %s/blobs/%s", response.Status, repository, digest) + } + // Read from the header rather than ContentLength: a HEAD's ContentLength is what the response + // says it would have sent, which Go reports faithfully, but a proxy in between is free to drop + // it, and the header is what the registry itself wrote. + if length := response.Header.Get("Content-Length"); length != "" { + if n, err := strconv.ParseInt(length, 10, 64); err == nil && n >= 0 { + return n, nil + } + } + if response.ContentLength >= 0 { + return response.ContentLength, nil + } + return 0, fmt.Errorf("the artifact store holds %s/blobs/%s and will not say how large it is", repository, digest) +} + +// putBlob uploads a small blob unless the repository already has it: ask where, then put it there +// naming the digest — the registry's own two steps, the same the builder takes for an archive. +func (s Store) putBlob(ctx context.Context, repository, digest string, body []byte) error { + if there, err := s.has(ctx, s.url(repository, "blobs", digest)); err != nil { + return err + } else if there { + return nil + } + start, err := http.NewRequestWithContext(ctx, http.MethodPost, + "http://"+s.Address+"/v2/"+repository+"/blobs/uploads/", nil) + if err != nil { + return err + } + begun, err := s.client().Do(start) + if err != nil { + return fmt.Errorf("cannot start an upload to %s: %w", repository, err) + } + begun.Body.Close() + if begun.StatusCode != http.StatusAccepted { + return fmt.Errorf("the artifact store answered %s when asked where to put a blob in %s", begun.Status, repository) + } + where := begun.Header.Get("Location") + if where == "" { + return fmt.Errorf("the artifact store accepted an upload to %s and said nowhere to put it", repository) + } + if strings.HasPrefix(where, "/") { + where = "http://" + s.Address + where + } + separator := "?" + if strings.Contains(where, "?") { + separator = "&" + } + put, err := http.NewRequestWithContext(ctx, http.MethodPut, where+separator+"digest="+digest, bytes.NewReader(body)) + if err != nil { + return err + } + put.Header.Set("Content-Type", "application/octet-stream") + done, err := s.client().Do(put) + if err != nil { + return err + } + defer done.Body.Close() + if done.StatusCode != http.StatusCreated { + said, _ := io.ReadAll(io.LimitReader(done.Body, 4096)) + return fmt.Errorf("the artifact store refused a blob in %s: %s %s", repository, done.Status, strings.TrimSpace(string(said))) + } + return nil +} + +// has is whether the store answers 200 for a HEAD at that URL. +func (s Store) has(ctx context.Context, url string, accept ...string) (bool, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil) + if err != nil { + return false, err + } + for _, media := range accept { + request.Header.Add("Accept", media) + } + response, err := s.client().Do(request) + if err != nil { + return false, fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err) + } + defer response.Body.Close() + switch response.StatusCode { + case http.StatusOK: + return true, nil + case http.StatusNotFound: + return false, nil + default: + return false, fmt.Errorf("the artifact store answered %s for %s", response.Status, url) + } +} + +func (s Store) url(repository, kind, digest string) string { + return "http://" + s.Address + "/v2/" + repository + "/" + kind + "/" + digest +} + +func (s Store) client() *http.Client { + if s.HTTP != nil { + return s.HTTP + } + return &http.Client{Timeout: 30 * time.Second} +} diff --git a/modules/artifact-store-tools/internal/registry/hold_test.go b/modules/artifact-store-tools/internal/registry/hold_test.go new file mode 100644 index 00000000..8a3a16fa --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/hold_test.go @@ -0,0 +1,268 @@ +package registry + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Every kept archive is held by a manifest (novox/hq issue 253, ADR 0189). +// +// Against an in-memory registry that keeps blobs and manifests per repository and refuses what a +// registry refuses — a blob whose digest does not match, a manifest whose digest does not match +// or whose blobs the repository does not have, a manifest asked for without an Accept naming its +// type. What is asserted is this side's decisions; the live test below asserts the registry's. + +type memRegistry struct { + mu sync.Mutex + blobs map[string][]byte // repository + "@" + digest + manifests map[string][]byte // repository + "@" + digest + writes []string // every PUT and DELETE, as "METHOD path" +} + +func digestOf(body []byte) string { + sum := sha256.Sum256(body) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func (m *memRegistry) serve(t *testing.T) Store { + t.Helper() + m.blobs = map[string][]byte{} + m.manifests = map[string][]byte{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + m.mu.Lock() + defer m.mu.Unlock() + path := strings.TrimPrefix(r.URL.Path, "/v2/") + if r.Method == http.MethodPut || r.Method == http.MethodDelete { + m.writes = append(m.writes, r.Method+" "+r.URL.Path) + } + switch { + case r.Method == http.MethodPost && strings.HasSuffix(path, "/blobs/uploads/"): + repository := strings.TrimSuffix(path, "/blobs/uploads/") + w.Header().Set("Location", "/upload/"+repository+"?state=x") + w.WriteHeader(http.StatusAccepted) + case r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/upload/"): + repository := strings.TrimPrefix(r.URL.Path, "/upload/") + body, _ := io.ReadAll(r.Body) + digest := r.URL.Query().Get("digest") + if digest != digestOf(body) { + w.WriteHeader(http.StatusBadRequest) + return + } + m.blobs[repository+"@"+digest] = body + w.WriteHeader(http.StatusCreated) + case strings.Contains(path, "/blobs/"): + repository, digest, _ := strings.Cut(path, "/blobs/") + key := repository + "@" + digest + body, ok := m.blobs[key] + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + switch r.Method { + case http.MethodHead: + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.WriteHeader(http.StatusOK) + case http.MethodDelete: + delete(m.blobs, key) + w.WriteHeader(http.StatusAccepted) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } + case strings.Contains(path, "/manifests/"): + repository, digest, _ := strings.Cut(path, "/manifests/") + key := repository + "@" + digest + switch r.Method { + case http.MethodHead: + if _, ok := m.manifests[key]; !ok || !strings.Contains(r.Header.Get("Accept"), mediaManifest) { + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + case http.MethodPut: + body, _ := io.ReadAll(r.Body) + if digest != digestOf(body) { + w.WriteHeader(http.StatusBadRequest) + return + } + var named holderManifest + if err := json.Unmarshal(body, &named); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + for _, d := range append([]descriptor{named.Config}, named.Layers...) { + if _, ok := m.blobs[repository+"@"+d.Digest]; !ok { + w.WriteHeader(http.StatusBadRequest) + fmt.Fprintf(w, "MANIFEST_BLOB_UNKNOWN %s", d.Digest) + return + } + } + m.manifests[key] = body + w.WriteHeader(http.StatusCreated) + case http.MethodDelete: + if _, ok := m.manifests[key]; !ok { + w.WriteHeader(http.StatusNotFound) + return + } + delete(m.manifests, key) + w.WriteHeader(http.StatusAccepted) + } + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(server.Close) + return Store{Address: strings.TrimPrefix(server.URL, "http://")} +} + +// bare puts an archive in the store the way the builder did before holders: a blob, nothing more. +func (m *memRegistry) bare(repository string, body []byte) string { + m.mu.Lock() + defer m.mu.Unlock() + digest := digestOf(body) + m.blobs[repository+"@"+digest] = body + return catalogue.ArtifactStoreScheme + repository + "/blobs/" + digest +} + +func TestTheHolderIsComposedFromTheDigestAndSizeAlone(t *testing.T) { + // The sweep must arrive at the very manifest the builder wrote, with nothing recorded between + // them. Same inputs, same bytes — and a different size is a different holder, so a holder can + // never be mistaken for one of a different blob. + digest := "sha256:" + strings.Repeat("a", 64) + one, first := Holder(digest, 42) + two, second := Holder(digest, 42) + if !bytes.Equal(one, two) || first != second { + t.Fatalf("the same archive composed two holders:\n%s\n%s", one, two) + } + if _, other := Holder(digest, 43); other == first { + t.Fatal("a different size composed the same holder") + } + want := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json",` + + `"config":{"mediaType":"application/vnd.oci.empty.v1+json",` + + `"digest":"sha256:44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a","size":2},` + + `"layers":[{"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip","digest":"` + digest + `","size":42}]}` + if string(one) != want { + t.Fatalf("the holder is\n%s\nwant\n%s", one, want) + } + if digestOf(emptyConfig) != emptyDigest { + t.Fatalf("the empty config's digest is %s, not %s", digestOf(emptyConfig), emptyDigest) + } +} + +func TestHoldBackfillsABareArchiveAndIsIdempotent(t *testing.T) { + // The archives published before this have no holder. Hold, run over every kept archive on + // every sweep, writes one the first time and nothing after. + m := &memRegistry{} + store := m.serve(t) + ctx := context.Background() + body := []byte("a theme") + reference := m.bare("shell/config", body) + + if held, err := store.Held(ctx, reference); err != nil || held { + t.Fatalf("a bare blob reads as held=%v (%v)", held, err) + } + wrote, err := store.Hold(ctx, reference) + if err != nil { + t.Fatal(err) + } + if !wrote { + t.Fatal("holding a bare archive wrote nothing") + } + _, holder := Holder(digestOf(body), int64(len(body))) + if _, ok := m.manifests["shell/config@"+holder]; !ok { + t.Fatalf("the store holds manifests %v; want %s", m.manifests, holder) + } + if held, err := store.Held(ctx, reference); err != nil || !held { + t.Fatalf("after holding, held=%v (%v)", held, err) + } + + writes := len(m.writes) + wrote, err = store.Hold(ctx, reference) + if err != nil { + t.Fatal(err) + } + if wrote || len(m.writes) != writes { + t.Fatalf("holding again wrote %v", m.writes[writes:]) + } +} + +func TestAnImageNeedsNoHolderAndAMissingArchiveIsGone(t *testing.T) { + m := &memRegistry{} + store := m.serve(t) + ctx := context.Background() + + // An image is its own manifest: nothing is asked. + wrote, err := store.Hold(ctx, catalogue.ArtifactStoreScheme+"web/app@sha256:"+strings.Repeat("b", 64)) + if err != nil || wrote || len(m.writes) != 0 { + t.Fatalf("holding an image wrote=%v err=%v writes=%v", wrote, err, m.writes) + } + // An archive the store does not have is a fact to report, not something to invent a holder for. + _, err = store.Hold(ctx, catalogue.ArtifactStoreScheme+"web/config/blobs/sha256:"+strings.Repeat("c", 64)) + if !errors.Is(err, Gone) { + t.Fatalf("holding a missing archive answered %v, want Gone", err) + } + // And a reference that is not the mesh's is refused as such. + if _, err := store.Hold(ctx, "docker.io/library/registry@sha256:abc"); !errors.Is(err, ErrNotOurs) { + t.Fatalf("holding a vendor's image answered %v, want ErrNotOurs", err) + } +} + +func TestLettingGoOfAnArchiveDeletesItsHolderFirst(t *testing.T) { + // A holder left behind would keep the bytes through every collection while the record said + // collected; the link deleted first and the holder failing after would be that exactly. + m := &memRegistry{} + store := m.serve(t) + ctx := context.Background() + body := []byte("an old theme") + reference := m.bare("shell/config", body) + if _, err := store.Hold(ctx, reference); err != nil { + t.Fatal(err) + } + m.writes = nil + + if err := store.LetGo(ctx, reference); err != nil { + t.Fatal(err) + } + _, holder := Holder(digestOf(body), int64(len(body))) + want := []string{ + "DELETE /v2/shell/config/manifests/" + holder, + "DELETE /v2/shell/config/blobs/" + digestOf(body), + } + if strings.Join(m.writes, "\n") != strings.Join(want, "\n") { + t.Fatalf("the store was asked\n%s\nwant\n%s", strings.Join(m.writes, "\n"), strings.Join(want, "\n")) + } + if len(m.manifests) != 0 { + t.Fatalf("a holder survived: %v", m.manifests) + } + // Asked again, the archive is already gone, which is the outcome wanted. + if err := store.LetGo(ctx, reference); !errors.Is(err, Gone) { + t.Fatalf("letting go twice answered %v, want Gone", err) + } +} + +func TestLettingGoOfAnUnheldArchiveStillDeletesIt(t *testing.T) { + // An archive published before holders and let go of before any sweep held it: the holder's + // delete answers 404, which is the outcome wanted, and the blob still goes. + m := &memRegistry{} + store := m.serve(t) + reference := m.bare("shell/config", []byte("never held")) + if err := store.LetGo(context.Background(), reference); err != nil { + t.Fatal(err) + } + if len(m.blobs) != 0 { + t.Fatalf("the blob survived: %v", m.blobs) + } +} diff --git a/modules/artifact-store-tools/internal/registry/platforms_test.go b/modules/artifact-store-tools/internal/registry/platforms_test.go new file mode 100644 index 00000000..a335fb2c --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/platforms_test.go @@ -0,0 +1,176 @@ +package registry + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// An index is let go of with the platform manifests it names, except those a kept index of the same +// repository names too (novox/hq ADR 0257). Each platform manifest is a manifest of the repository in +// its own right, and the store's collector keeps every manifest a repository holds — so an index let go +// of alone frees nothing of the images it names. + +func digestN(n int) string { return fmt.Sprintf("sha256:%064x", n) } + +// indexStore holds indexes and images in one repository, answers GETs with their documents, and records +// every delete. +type indexStore struct { + mu sync.Mutex + indexes map[string][]string // digest → the platform manifests it names + images map[string]bool + deleted []string +} + +func (s *indexStore) serve(t *testing.T) Store { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + defer s.mu.Unlock() + digest := r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:] + switch r.Method { + case http.MethodGet: + if children, ok := s.indexes[digest]; ok { + var named []string + for _, c := range children { + named = append(named, `{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"`+c+`","size":1}`) + } + w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json") + _, _ = w.Write([]byte(`{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[` + + strings.Join(named, ",") + `]}`)) + return + } + if s.images[digest] { + w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json") + _, _ = w.Write([]byte(`{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","layers":[]}`)) + return + } + w.WriteHeader(http.StatusNotFound) + case http.MethodHead: + if _, ok := s.indexes[digest]; ok || s.images[digest] { + w.WriteHeader(http.StatusOK) + return + } + w.WriteHeader(http.StatusNotFound) + case http.MethodDelete: + if _, ok := s.indexes[digest]; !ok && !s.images[digest] { + w.WriteHeader(http.StatusNotFound) + return + } + s.deleted = append(s.deleted, digest) + delete(s.indexes, digest) + delete(s.images, digest) + w.WriteHeader(http.StatusAccepted) + default: + w.WriteHeader(http.StatusBadRequest) + } + })) + t.Cleanup(server.Close) + return Store{Address: strings.TrimPrefix(server.URL, "http://")} +} + +func TestAnIndexGoesWithItsPlatformsAndAKeptIndexKeepsItsOwn(t *testing.T) { + // Two indexes of one image's repository: the old one names platforms 11 and 12, the kept one 12 and 13. + s := &indexStore{ + indexes: map[string][]string{digestN(1): {digestN(11), digestN(12)}, digestN(2): {digestN(12), digestN(13)}}, + images: map[string]bool{digestN(11): true, digestN(12): true, digestN(13): true}, + } + store := s.serve(t) + repository := "upstream/docker.io/library/golang" + old := catalogue.ArtifactStoreScheme + repository + "@" + digestN(1) + kept := catalogue.ArtifactStoreScheme + repository + "@" + digestN(2) + spare, err := store.SpareKeptIndexes(context.Background(), []string{kept}, []string{old}) + if err != nil { + t.Fatal(err) + } + store.Spare = spare + + if err := store.LetGo(context.Background(), old); err != nil { + t.Fatal(err) + } + // Its own platform first, the index last; the platform the kept index names is left in place. + if !slices.Equal(s.deleted, []string{digestN(11), digestN(1)}) { + t.Fatalf("deleted %v; want the old index's own platform, then the index", s.deleted) + } + if !s.images[digestN(12)] || !s.images[digestN(13)] { + t.Fatal("a platform a kept index names was let go of") + } +} + +func TestWithoutSparingAnIndexGoesAloneAndAnImageNamesNoPlatforms(t *testing.T) { + s := &indexStore{indexes: map[string][]string{digestN(1): {digestN(11)}}, images: map[string]bool{digestN(11): true, digestN(5): true}} + store := s.serve(t) + // No Spare: as before ADR 0257, the index alone. + if err := store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/base@"+digestN(1)); err != nil { + t.Fatal(err) + } + if !slices.Equal(s.deleted, []string{digestN(1)}) { + t.Fatalf("without a Spare, deleted %v", s.deleted) + } + // An image is one delete, with a Spare or without. + s.deleted = nil + spare, err := store.SpareKeptIndexes(context.Background(), nil, []string{catalogue.ArtifactStoreScheme + "web/app@" + digestN(5)}) + if err != nil { + t.Fatal(err) + } + store.Spare = spare + if err := store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/app@"+digestN(5)); err != nil { + t.Fatal(err) + } + if !slices.Equal(s.deleted, []string{digestN(5)}) { + t.Fatalf("an image was let go of as %v", s.deleted) + } + // An index the store no longer holds is Gone, and nothing is deleted on its account. + s.deleted = nil + err = store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/base@"+digestN(9)) + if !errors.Is(err, Gone) || len(s.deleted) != 0 { + t.Fatalf("an index the store does not hold: %v, deleted %v", err, s.deleted) + } +} + +func TestASpareThatCannotBeReadKeepsTheIndex(t *testing.T) { + s := &indexStore{indexes: map[string][]string{digestN(1): {digestN(11)}}, images: map[string]bool{digestN(11): true}} + store := s.serve(t) + store.Spare = func(context.Context, string) (map[string]bool, error) { return nil, fmt.Errorf("the store went away") } + err := store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/base@"+digestN(1)) + if err == nil || !strings.Contains(err.Error(), "was kept") { + t.Fatalf("a spare that could not be read: %v", err) + } + if len(s.deleted) != 0 { + t.Fatalf("deleted %v without knowing what a kept index names", s.deleted) + } +} + +// The spare list is read whole before the sweep: a kept index that cannot be read is an error before +// anything is deleted, and a repository not read before keeps its index. +func TestASpareListIsReadWholeBeforeAnythingGoes(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + t.Cleanup(server.Close) + store := Store{Address: strings.TrimPrefix(server.URL, "http://")} + kept := catalogue.ArtifactStoreScheme + "web/base@" + digestN(2) + old := catalogue.ArtifactStoreScheme + "web/base@" + digestN(1) + if _, err := store.SpareKeptIndexes(context.Background(), []string{kept}, []string{old}); err == nil { + t.Fatal("a kept index the store would not answer for was taken as naming nothing") + } + + s := &indexStore{indexes: map[string][]string{digestN(1): {digestN(11)}}, images: map[string]bool{digestN(11): true}} + good := s.serve(t) + spare, err := good.SpareKeptIndexes(context.Background(), nil, []string{catalogue.ArtifactStoreScheme + "other/repo@" + digestN(3)}) + if err != nil { + t.Fatal(err) + } + good.Spare = spare + if err := good.LetGo(context.Background(), old); err == nil || len(s.deleted) != 0 { + t.Fatalf("an index in a repository not read before the sweep was let go of: %v, deleted %v", err, s.deleted) + } +} diff --git a/modules/artifact-store-tools/internal/registry/store.go b/modules/artifact-store-tools/internal/registry/store.go new file mode 100644 index 00000000..df2503c3 --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/store.go @@ -0,0 +1,325 @@ +// Package registry speaks to the mesh's artifact store over its own door. +// +// Only what the mesh needs that nothing else does: letting go of something it put there +// (novox/hq ADR 0189, issue 108), and holding every archive it keeps by a manifest so the store's +// own collector does not take it (novox/hq issue 253). Pushing is the builder's, through the container runtime; reading +// is every machine's, through its runtime. This is the one operation that belongs to the thing +// holding the records, because it is the only one that is a decision rather than a transfer. +// +// Copied from mesh-controller internal/artifacts at e04846e9 (novox/hq ADR 0571 clause 2): the seat artifact-store lets +// go of what it no longer keeps exactly as the controller's sweep did; the controller's copy goes with its lease. +package registry + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Store is the artifact store at an address, as this machine reaches it. +type Store struct { + // Address is `host:port` — the store as the caller reaches it now, composed and never + // recorded (novox/hq 04-ISSUES/102). + Address string + // HTTP is the client used; nil is a client with a modest timeout. + HTTP *http.Client + // Spare says, for one repository, which manifests a kept index there names: the platform + // manifests an index being let go of must leave in place (novox/hq ADR 0257). Nil spares every + // platform manifest — an index is then let go of alone, as it was before. + Spare func(ctx context.Context, repository string) (map[string]bool, error) +} + +// Gone is the answer when the store does not hold it: the outcome wanted, already true. +var Gone = errors.New("the store does not hold it") + +// ErrNotOurs is a reference this sweep will not address: not the mesh's own, or naming nothing +// the store holds by digest. +// +// **A fact about the record, not about the store** (novox/hq issue 226). The two deserve opposite +// responses — skip one and go on, abandon the sweep for the other — and collapsing them into "an +// error" is how a cautious loop became one that did nothing while reporting the right number. +var ErrNotOurs = errors.New("not a reference into the mesh's artifact store") + +// LetGo asks the store to drop one artifact the mesh recorded making. +// +// Takes a reference as the mesh records it — `artifact-store:///@sha256:…` for +// an image, `…/blobs/sha256:…` for an archive — because that is the identity every record uses, +// and composes the address here at the moment of use. +// +// An archive is let go of in two deletes, its holder manifest and then the blob's link (novox/hq +// issue 253); an image in one. +// +// Returns Gone when the store answers that it does not have it. That is not a failure: the sweep +// wants the artifact absent, and it is. It is distinguished from success only so a caller can say +// which of the two happened. +func (s Store) LetGo(ctx context.Context, reference string) error { + // **Strict, and deliberately** (novox/hq issue 226). Only a reference the mesh keeps in its + // own vocabulary is addressed here. `Recorded` would read `docker.io/library/registry@sha256:…` + // as the mesh's too — it cannot tell one registry host from another — so normalising belongs + // where the provenance is known, which is the sweep reading its own build records, not here + // where the only job is to refuse anything that is not plainly ours. + path, kept := catalogue.InArtifactStore(reference) + if !kept { + // Nothing the mesh put in its own store. Refused rather than attempted: composing a + // delete for a reference of unknown shape is how a sweep reaches something that is not + // the mesh's. Distinguished from a store that refuses, so a sweep skips this and goes on. + return fmt.Errorf("%w: %s", ErrNotOurs, reference) + } + if s.Address == "" { + return fmt.Errorf("this mesh has no artifact store on its network to ask about %s", reference) + } + repository, kind, digest, err := split(path) + if err != nil { + return err + } + if kind == "blobs" { + // **An archive's holder goes before the archive** (novox/hq issue 253): a manifest left + // naming the blob would keep its bytes through every collection while the record said + // collected. Gone here means the store has no such blob, so there is nothing to let go. + if err := s.letGoOfHolder(ctx, repository, digest); err != nil { + return err + } + } + if kind == "manifests" && s.Spare != nil { + // **An index's platform manifests go before the index** (novox/hq ADR 0257). Each was put + // under its own digest, so each is a manifest of the repository in its own right, and the + // store's collector keeps every manifest a repository holds: an index let go of alone frees + // no byte of the images it names. Before, not after, so a sweep cut short leaves the index + // to be asked about again, and with it the list of what is still to go. + if err := s.letGoOfPlatforms(ctx, repository, digest, reference); err != nil { + return err + } + } + return s.remove(ctx, s.url(repository, kind, digest), reference) +} + +// letGoOfPlatforms deletes the manifests an index names, except those a kept index of the same +// repository also names. A manifest that is not an index names none. +func (s Store) letGoOfPlatforms(ctx context.Context, repository, digest, reference string) error { + children, err := s.Platforms(ctx, repository, digest) + if err != nil || len(children) == 0 { + // Gone: there is no index to read, and the delete that follows says so. + if errors.Is(err, Gone) { + return nil + } + return err + } + spared, err := s.Spare(ctx, repository) + if err != nil { + return fmt.Errorf("cannot say which platform manifests of %s a kept index names, so %s was kept: %w", + repository, reference, err) + } + for _, child := range children { + if spared[child] { + continue + } + if err := s.remove(ctx, s.url(repository, "manifests", child), reference+" (its "+child+")"); err != nil && !errors.Is(err, Gone) { + return err + } + } + return nil +} + +// anyManifest is every manifest the store may hold, indexes included, so it answers with the +// document as it is. +var anyManifest = []string{ + "application/vnd.oci.image.index.v1+json", + "application/vnd.docker.distribution.manifest.list.v2+json", + "application/vnd.oci.image.manifest.v1+json", + "application/vnd.docker.distribution.manifest.v2+json", +} + +// Platforms is the manifests an index names, by digest; none for a manifest that is not an index. +// Gone when the store does not hold it. +func (s Store) Platforms(ctx context.Context, repository, digest string) ([]string, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url(repository, "manifests", digest), nil) + if err != nil { + return nil, err + } + request.Header.Set("Accept", strings.Join(anyManifest, ", ")) + response, err := s.client().Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + switch response.StatusCode { + case http.StatusOK: + case http.StatusNotFound: + return nil, Gone + default: + return nil, fmt.Errorf("the artifact store answered %s when asked for %s@%s", response.Status, repository, digest) + } + var document struct { + Manifests []struct { + Digest string `json:"digest"` + } `json:"manifests"` + } + if err := json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&document); err != nil { + return nil, fmt.Errorf("%s@%s is not a manifest: %w", repository, digest, err) + } + var out []string + for _, m := range document.Manifests { + if strings.HasPrefix(m.Digest, "sha256:") { + out = append(out, m.Digest) + } + } + return out, nil +} + +// SpareKeptIndexes is a Spare for a sweep, read whole before it starts: for every repository an +// image reference of `touching` is in, each kept manifest there and every platform manifest a kept +// index there names. Any read that fails is an error, and the sweep lets nothing go. A repository not +// read here is answered with an error, so an index in it is kept. A kept index the store no longer +// holds names nothing. +func (s Store) SpareKeptIndexes(ctx context.Context, kept, touching []string) (func(context.Context, string) (map[string]bool, error), error) { + imageIn := func(reference string) (repository, digest string, ok bool) { + path, ours := catalogue.InArtifactStore(reference) + if !ours { + return "", "", false + } + repository, kind, digest, err := split(path) + if err != nil || kind != "manifests" { + return "", "", false + } + return repository, digest, true + } + read := map[string]map[string]bool{} + for _, reference := range touching { + if repository, _, ok := imageIn(reference); ok { + read[repository] = map[string]bool{} + } + } + for _, reference := range kept { + repository, digest, ok := imageIn(reference) + spared, touched := read[repository] + if !ok || !touched { + continue + } + spared[digest] = true + children, err := s.Platforms(ctx, repository, digest) + if errors.Is(err, Gone) { + continue + } + if err != nil { + return nil, fmt.Errorf("reading the kept %s@%s: %w", repository, digest, err) + } + for _, c := range children { + spared[c] = true + } + } + return func(_ context.Context, repository string) (map[string]bool, error) { + spared, done := read[repository] + if !done { + return nil, fmt.Errorf("%s was not read before the sweep began", repository) + } + return spared, nil + }, nil +} + +// remove asks the store to delete what is at url. Gone when it has no such thing. +func (s Store) remove(ctx context.Context, url, what string) error { + request, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil) + if err != nil { + return err + } + response, err := s.client().Do(request) + if err != nil { + return err + } + defer response.Body.Close() + switch response.StatusCode { + case http.StatusAccepted, http.StatusOK, http.StatusNoContent: + return nil + case http.StatusNotFound: + return Gone + case http.StatusMethodNotAllowed: + // The registry was started without deletion enabled. Said plainly, because the remedy is + // a setting on the store's module and not anything about this artifact. + return fmt.Errorf( + "the artifact store refuses deletion: its server was started without it enabled "+ + "(REGISTRY_STORAGE_DELETE_ENABLED), so nothing can be collected until the store "+ + "module is applied again (novox/hq ADR 0189). Asking about %s", what) + default: + return fmt.Errorf("the artifact store answered %s for %s", response.Status, what) + } +} + +// split reads a recorded path into the repository, which endpoint names the thing, and the digest. +// +// Two shapes, which are the two the mesh records: `@sha256:` is a manifest, and +// `/blobs/sha256:` is a blob. +func split(path string) (repository, kind, digest string, err error) { + if before, after, ok := strings.Cut(path, "@sha256:"); ok { + return before, "manifests", "sha256:" + after, nil + } + if before, after, ok := strings.Cut(path, "/blobs/sha256:"); ok { + return before, "blobs", "sha256:" + after, nil + } + return "", "", "", fmt.Errorf("%w: %q names nothing the store holds by digest", ErrNotOurs, path) +} + +// IsIndex is whether a recorded reference is an index the store holds: one that names manifests. A +// blob, or a manifest that names none, is not. Gone when the store does not hold it. +func (s Store) IsIndex(ctx context.Context, reference string) (bool, error) { + path, ours := catalogue.InArtifactStore(reference) + if !ours { + return false, fmt.Errorf("%w: %s", ErrNotOurs, reference) + } + repository, kind, digest, err := split(path) + if err != nil || kind != "manifests" { + return false, err + } + children, err := s.Platforms(ctx, repository, digest) + return len(children) > 0, err +} + +// HoldsManifest is whether the store holds the manifest a recorded image reference names. +func (s Store) HoldsManifest(ctx context.Context, reference string) (bool, error) { + path, ours := catalogue.InArtifactStore(reference) + if !ours { + return false, fmt.Errorf("%w: %s", ErrNotOurs, reference) + } + repository, kind, digest, err := split(path) + if err != nil { + return false, err + } + if kind != "manifests" { + return false, fmt.Errorf("%w: %s names a blob, not a manifest", ErrNotOurs, reference) + } + return s.has(ctx, s.url(repository, kind, digest), anyManifest...) +} + +// Holds is whether the store holds what a recorded reference names: the blob of an archive, the manifest of an image. +// Gone when it does not; asks and changes nothing (ADR 0526 clause 3: a registration by hand is refused at bytes that +// are not there). +func (s Store) Holds(ctx context.Context, reference string) error { + path, ours := catalogue.InArtifactStore(reference) + if !ours { + return fmt.Errorf("%w: %s", ErrNotOurs, reference) + } + repository, kind, digest, err := split(path) + if err != nil { + return err + } + if s.Address == "" { + return fmt.Errorf("this mesh has no artifact store on its network to ask about %s", reference) + } + if kind == "blobs" { + _, err := s.blobSize(ctx, repository, digest) + return err + } + held, err := s.has(ctx, s.url(repository, kind, digest), anyManifest...) + if err != nil { + return err + } + if !held { + return Gone + } + return nil +} diff --git a/modules/artifact-store-tools/internal/registry/store_test.go b/modules/artifact-store-tools/internal/registry/store_test.go new file mode 100644 index 00000000..334b95a5 --- /dev/null +++ b/modules/artifact-store-tools/internal/registry/store_test.go @@ -0,0 +1,134 @@ +package registry + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Asking the store to let go of what the mesh no longer keeps (novox/hq ADR 0189, issue 108). +// +// A fake store records what it was asked to delete, so what is asserted is the mesh's decision +// and the shape of the request — not the registry's behaviour, which is the registry's to test. + +func fakeStore(t *testing.T, answer int) (Store, *[]string) { + t.Helper() + var asked []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/blobs/") { + // An archive's size, asked so its holder can be named (novox/hq issue 253). A store + // that does not have the thing does not have its blob either. + if answer == http.StatusNotFound { + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Length", "7") + w.WriteHeader(http.StatusOK) + return + } + if r.Method != http.MethodDelete { + t.Errorf("the store was asked %s %s; collecting is a delete", r.Method, r.URL.Path) + } + asked = append(asked, r.URL.Path) + w.WriteHeader(answer) + })) + t.Cleanup(server.Close) + return Store{Address: strings.TrimPrefix(server.URL, "http://")}, &asked +} + +func TestAnImageAndAnArchiveAreAskedForAtTheirOwnEndpoints(t *testing.T) { + // The two shapes the mesh records: a manifest by digest, and a blob by digest. They are + // different endpoints, and asking at the wrong one answers 404 — which this would then + // record as collected, leaving the bytes on disk for ever while the record says otherwise. + store, asked := fakeStore(t, http.StatusAccepted) + ctx := context.Background() + + image := catalogue.ArtifactStoreScheme + "web/app@sha256:abc123" + archive := catalogue.ArtifactStoreScheme + "web/config/blobs/sha256:def456" + if err := store.LetGo(ctx, image); err != nil { + t.Fatal(err) + } + if err := store.LetGo(ctx, archive); err != nil { + t.Fatal(err) + } + // The archive's holder goes first, then the archive (novox/hq issue 253). + _, holder := Holder("sha256:def456", 7) + want := []string{ + "/v2/web/app/manifests/sha256:abc123", + "/v2/web/config/manifests/" + holder, + "/v2/web/config/blobs/sha256:def456", + } + if strings.Join(*asked, " ") != strings.Join(want, " ") { + t.Fatalf("the store was asked %v; want %v", *asked, want) + } +} + +func TestAStoreThatDoesNotHaveItAnswersGone(t *testing.T) { + // The outcome wanted, already true. Told apart from success only so the sweep can say which + // happened; both are recorded, because retrying for ever is the thing to avoid. + store, _ := fakeStore(t, http.StatusNotFound) + err := store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/app@sha256:abc123") + if !errors.Is(err, Gone) { + t.Fatalf("a store that does not hold it answered %v, want Gone", err) + } +} + +func TestAStoreWithDeletionOffSaysSoAndNamesTheRemedy(t *testing.T) { + // The registry answers 405 when it was started without deletion enabled. The remedy is a + // setting on the store's module, and saying "405" would send somebody to the wrong place. + store, _ := fakeStore(t, http.StatusMethodNotAllowed) + err := store.LetGo(context.Background(), catalogue.ArtifactStoreScheme+"web/app@sha256:abc123") + if err == nil { + t.Fatal("a store that refuses deletion was read as success") + } + if !strings.Contains(err.Error(), "REGISTRY_STORAGE_DELETE_ENABLED") { + t.Fatalf("the refusal does not name the remedy: %v", err) + } +} + +func TestAReferenceThatIsNotTheMeshsOwnIsNeverAsked(t *testing.T) { + // The whole safety of the sweep is that it names only what the mesh recorded putting there. + // A reference of another shape — a vendor's image, a package version — is refused rather + // than composed into a delete somewhere that is not the mesh's store. + store, asked := fakeStore(t, http.StatusAccepted) + for _, reference := range []string{ + "docker.io/library/registry@sha256:abc123", + "registry@sha256:abc123", + "1.4.2", + } { + if err := store.LetGo(context.Background(), reference); err == nil { + t.Errorf("%s was asked about; it is not a reference into the mesh's store", reference) + } + } + if len(*asked) != 0 { + t.Fatalf("the store was asked about %v", *asked) + } +} + +// A reference this sweep will not address says so as ErrNotOurs, which is a fact about the +// record and not about the store (novox/hq issue 226). +// +// The sweep skips one and abandons itself for the other, so they cannot be the same error. The +// first live run met a reference recorded with the store's old address, read the refusal as "the +// store refuses everything", and collected none of the 1681 it had found. +func TestAReferenceThisSweepWillNotAddressIsToldApartFromAStoreRefusing(t *testing.T) { + store, asked := fakeStore(t, http.StatusAccepted) + for _, reference := range []string{ + "docker.io/library/registry@sha256:abc123", + "127.0.0.1:5100/mesh-tools/build@sha256:abc123", + "1.4.2", + } { + err := store.LetGo(context.Background(), reference) + if !errors.Is(err, ErrNotOurs) { + t.Errorf("%s answered %v; a sweep must be able to skip it and go on", reference, err) + } + } + if len(*asked) != 0 { + t.Fatalf("the store was asked about %v", *asked) + } +} diff --git a/modules/artifact-store-tools/module.json b/modules/artifact-store-tools/module.json index 208329cd..bf70046b 100644 --- a/modules/artifact-store-tools/module.json +++ b/modules/artifact-store-tools/module.json @@ -1,26 +1,50 @@ { "module": "artifact-store-tools", "version": "1", + "claims": [ + { + "name": "artifact-store", + "scope": "mesh", + "serves": [ + "artifacts", + "held", + "collect", + "mirrors", + "images" + ] + } + ], + "requires": [ + "artifact-store" + ], "invokes": [ - "seat:mesh-controller.artifacts", - "seat:mesh-controller.collect", + "seat:mesh-hand-acts.hand-act", "mesh-catalog.catalog_modules", "mesh-catalog.catalog_module" ], + "reads": [ + "mesh-delivery-assigner.wants", + "mesh-delivery-builder.builds", + "node-engine.have", + "mesh-controller.lease" + ], + "state": [ + "findings" + ], "tools": [ "artifact_store_repositories", "artifact_store_usage", "artifact_store_references", "artifact_store_collect" ], - "config": { - "older-store-names": { - "type": "string", - "description": "The names the artifact store had before, as host names or host:port separated by spaces, or none when it had no other. artifact_store_references reads every definition under these as well as under the store's current name, so an image a definition pins under one is said to be named by a definition (novox/hq issue 323).", - "pattern": "none|[A-Za-z0-9.:-]+( [A-Za-z0-9.:-]+)*", - "default": "none" - } - }, + "config": { + "older-store-names": { + "type": "string", + "description": "The names the artifact store had before, as host names or host:port separated by spaces, or none when it had no other. artifact_store_references reads every definition under these as well as under the store's current name, so an image a definition pins under one is said to be named by a definition (novox/hq issue 323).", + "pattern": "none|[A-Za-z0-9.:-]+( [A-Za-z0-9.:-]+)*", + "default": "none" + } + }, "resources": [ { "id": "state", @@ -36,6 +60,14 @@ "mode": "0600", "trusted": false, "template": "files/store-names" + }, + { + "id": "store-address", + "type": "file", + "path": "${dir:state}/store-address", + "mode": "0600", + "trusted": false, + "template": "files/store-address" } ], "build": { @@ -52,7 +84,8 @@ ], "env": { "MESH_ARTIFACT_STORE_CONTAINER": "mesh-registry", - "MESH_ARTIFACT_STORE_NAMES_FILE": "${dir:state}/store-names" + "MESH_ARTIFACT_STORE_NAMES_FILE": "${dir:state}/store-names", + "MESH_ARTIFACT_STORE_ADDRESS_FILE": "${dir:state}/store-address" } }, {