diff --git a/modules/artifact-store-tools/README.md b/modules/artifact-store-tools/README.md index 5a38db8f..bce43f72 100644 --- a/modules/artifact-store-tools/README.md +++ b/modules/artifact-store-tools/README.md @@ -115,3 +115,13 @@ Any command other than the two read-only scripts fails the test. Each directory or archive this module declares says how it treats a file it did not write (ADR 0422, ADR 0416). - `${dir:state}`: everything in it is the program's; the mesh moves nothing out (ADR 0422 §2). + +### The sweep through the registry seats + +The sweep writes nothing to the store's door (CAT-AS1, ADR 0617 §2): it asks the registry seats the store's holder +serves. Each kept bundle or archive is held on `oci-artifact-registry` and each kept copy of an upstream base on +`container-image-registry` (`hold`, with why it is kept) before anything goes; each eligible one is released, then +removed for `retention` on its seat (`release`, `remove`). It reads a reference in both forms, `artifact-store://` +and `//@`, as one artifact, and says artifacts-gone in the form the build named it. A +call a seat refuses stops the sweep and writes the finding `sweep-refused` in the module's findings; the next sweep no +seat refuses clears it (`sweepseats.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 index 4335d786..2903caef 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go @@ -18,20 +18,12 @@ import ( // 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) } +// It is read only: what the sweep holds and removes it asks the registry seats (sweepseats.go). +func (d *door) Address() string { return d.s.Address } +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. diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go index ee0c9242..622fe3f1 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go @@ -69,15 +69,12 @@ type Bus interface { 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. +// Door is the artifact store's own door, read only: registry.Store, behind an interface so a test needs no store. What +// the sweep holds and removes it asks the registry seats (Seats, sweepseats.go). 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. @@ -87,8 +84,10 @@ type Holder struct { // 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) + // Seats asks a registry seat's verb, `seat:.` (hold, remove): the runtime's stdio.Ask in main. + Seats func(key string, body any) (json.RawMessage, error) + Now func() time.Time + Logf func(string, ...any) sweep sync.Mutex // one sweep at a time mu sync.Mutex @@ -217,7 +216,7 @@ func KeepSet(builds []Build, wants []Want) map[string]*Ref { out := map[string]*Ref{} for _, b := range builds { for _, a := range b.Artifacts { - reference := catalogue.Recorded(a.Reference) + reference := canonical(catalogue.Recorded(a.Reference)) if _, ours := catalogue.InArtifactStore(reference); !ours || a.Kind == "package" { continue } @@ -511,13 +510,13 @@ func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) return nil, err } a := &CollectAnswer{DryRun: s.DryRun, WouldLetGo: []string{}, LetGo: []string{}} - var keptArchives, keptImages, eligible []string + var keptArchives, eligible []string for _, r := range j.sorted() { switch { - case r.Kept && r.Kind == "archive": + case r.Kept && (r.Kind == "archive" || isCopy(r.Reference)): + // What retention would remove unless held: a bundle or an archive, and a copied base (ADR 0617 §2). 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 @@ -543,33 +542,25 @@ func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) 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. + // Hold before letting go (ADR 0251 §8, ADR 0617 §2): each kept archive and copied base is held on its seat; one that + // cannot be held stops the sweep before any removal and writes the sweep's finding. 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): + if held, known := inStore(j.view, r); known && !held { missing = append(missing, r) - case errors.Is(err, registry.ErrNotOurs): - default: + continue + } + if _, err := h.holdOnSeat(r, keptWhy(j.refs[r])); err != nil { 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) + h.sweepFinding("the hold of "+r, err) return a, nil } + a.HoldersWritten++ } 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 - } - } + refused := false for i, r := range eligible { if i >= s.Most || within.Err() != nil { a.Left = len(eligible) - i @@ -593,24 +584,19 @@ func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) 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"). + // Removed through its seat (ADR 0617 §2), 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.removeOnSeat(r, why); err != nil { + a.Stopped = fmt.Sprintf("a registry seat refused to remove %s, so nothing more was asked: %v", r, err) + a.Left = len(eligible) - i + h.sweepFinding("the removal of "+r, err) + refused = true + break + } 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) @@ -618,6 +604,9 @@ func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) a.LetGo = append(a.LetGo, r) } h.sayGone(j, a.LetGo, "let go of") + if !refused { + h.sweepClean() + } return a, nil } @@ -637,7 +626,7 @@ func (h *Holder) sayGone(j *judged, refs []string, how string) { byKey[b.Key] = g keys = append(keys, b.Key) } - g.Gone = append(g.Gone, r) + g.Gone = append(g.Gone, namedBy(b, r)) } } sort.Strings(keys) @@ -652,6 +641,17 @@ func (h *Holder) sayGone(j *judged, refs []string, how string) { } } +// namedBy is the reference b names r by, in the form its key holds (the older one or the seat's): the builder matches +// artifacts-gone against its own key. +func namedBy(b Build, r string) string { + for _, a := range b.Artifacts { + if rec := catalogue.Recorded(a.Reference); canonical(rec) == r { + return rec + } + } + return r +} + // AfterBuild is the sweep at a build's end. func (h *Holder) AfterBuild(key string) { a, err := h.Sweep(context.Background(), Sweeping{Most: MostPerSweep, By: "after the build " + key}) @@ -685,7 +685,7 @@ func (h *Holder) Held(refs []string) (delivery.HeldAnswer, error) { defer stop() a := delivery.HeldAnswer{Store: door.Address(), Held: []string{}, Gone: []string{}} for _, r := range refs { - switch err := door.Holds(asking, r); { + switch err := door.Holds(asking, canonical(r)); { case err == nil: a.Held = append(a.Held, r) case errors.Is(err, registry.Gone): @@ -899,6 +899,9 @@ func (h *Holder) Observe(ctx context.Context) error { return fmt.Errorf("the findings: %w", err) } for _, k := range keys { + if k == sweepFindingKey { + continue // the sweep's own finding, which the sweep clears + } if _, still := now[k]; !still { if err := h.Bus.Delete(FindingsState, k); err != nil { return err 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 index b97cfd9a..25f8b35d 100644 --- 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 @@ -173,15 +173,32 @@ func TestTheSeatSweepsAtABuildsEndOnARealBusWithNoLease(t *testing.T) { }); 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{Ask: func(key string, body any) (json.RawMessage, error) { + // The registry seats' holder on the bus (CAT-AS1): the sweep holds and removes through their verbs. + for _, seat := range []string{"oci-artifact-registry", "container-image-registry"} { + if _, err := nc.Subscribe("mesh.seat."+seat+".tool.*", func(m *nats.Msg) { + var body map[string]any + _ = json.Unmarshal(m.Data, &body) + mu.Lock() + out, err := door.seat("seat:"+seat+"."+m.Subject[strings.LastIndex(m.Subject, ".")+1:], body) + mu.Unlock() + if err != nil { + out, _ = json.Marshal(map[string]string{"error": err.Error()}) + } + _ = m.Respond(out) + }); err != nil { + t.Fatal(err) + } + } + 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 - }} + } + h := &Holder{Bus: findingsAt{realBus: bus}, Door: mem.Door, Listed: mem.Listed, Seats: ask, Now: time.Now, Logf: t.Logf} + h.Acts = handacts.Recorder{Ask: ask} // A build's end heard from the builds bucket, as follow hands it over. kv, _ := bus.kv(BuildsRead) 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 index eac183db..a0effbb3 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go @@ -3,7 +3,9 @@ package main import ( "context" "encoding/json" + "errors" "fmt" + "slices" "sort" "strings" "sync" @@ -15,6 +17,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "artifactstoretools/internal/registry" ) @@ -85,21 +89,58 @@ func (b *memBus) Publish(subject string, body any) error { return nil } -// memDoor is the store's door in memory: what it holds, which references are indexes, and what was asked. +// memDoor is the store's door in memory, read only, and the registry seats' holder: what it holds, which references are +// indexes, what the seats were asked (asked: "hold ", "let go "; calls: each seat call as its key and name), the +// holds asked (held), and the calls it refuses (refuse: by " "). type memDoor struct { holds map[string]bool indexes map[string]bool asked []string + calls []string + held map[string]bool + refuse map[string]error +} + +// seat is the registry seats' holder, as Holder.Seats calls it. +func (d *memDoor) seat(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + var in struct{ Name, Digest, Reason string } + _ = json.Unmarshal(raw, &in) + seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + ref := catalogue.ArtifactStoreScheme + in.Name + "@" + in.Digest + if seat == "oci-artifact-registry" { + ref = catalogue.ArtifactStoreScheme + in.Name + "/blobs/" + in.Digest + } + d.calls = append(d.calls, key+" "+in.Name+"@"+in.Digest+" "+in.Reason) + if err := d.refuse[verb+" "+ref]; err != nil { + return nil, err + } + if d.held == nil { + d.held = map[string]bool{} + } + switch verb { + case "hold": + d.asked = append(d.asked, "hold "+ref) + d.held[ref] = true + return json.Marshal(map[string]any{"held": true}) + case "release": + if !d.held[ref] { + return nil, fmt.Errorf("release: %s@%s is not held on %s", in.Name, in.Digest, seat) + } + delete(d.held, ref) + return json.Marshal(map[string]any{"released": true}) + case "remove": + if d.held[ref] { + return nil, fmt.Errorf("remove: refused: %s@%s is held; release it first", in.Name, in.Digest) + } + d.asked = append(d.asked, "let go "+ref) + delete(d.holds, ref) + return json.Marshal(map[string]any{"removed": true}) + } + return nil, fmt.Errorf("no verb %s", key) } 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 @@ -107,15 +148,6 @@ func (d *memDoor) Holds(_ context.Context, r string) error { 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 -} // mirror is a copy of an upstream base in the store, the n-th. func mirror(n int) string { @@ -194,7 +226,7 @@ func aMesh(t *testing.T) (*Holder, *memBus, *memDoor, *[]string) { want, _ := json.Marshal(delivery.Want{V: delivery.Version, FP: fp(1), Self: true}) _ = b.Put(WantsRead, "want.ace.letta", json.RawMessage(want)) var acts []string - h := &Holder{Bus: b, Door: func() (Door, error) { return door, nil }, + h := &Holder{Bus: b, Door: func() (Door, error) { return door, nil }, Seats: door.seat, 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) { @@ -246,18 +278,43 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T) left int skipped int unlog int + // held is how many kept archives and copies are held (7 when 0); seatForm names letta's 2nd and 3rd builds in the + // seat's form; refuse is the seat call refused, " ". + held int + seatForm bool + refuse string }{ {name: "a build's end", how: "build", letGo: all}, + // ADR 0605 §9: a build that names its artifacts in the seat's form is the same artifact: removed and held under the + // seat's name and digest, and said gone in its own form. + {name: "the seat's form", most: MostPerSweep, letGo: all, seatForm: true}, + // A seat that refuses a removal stops the sweep; the finding sweep-refused is written, and cleared by a clean run. + {name: "a refused removal", most: MostPerSweep, letGo: []string{archiveRef("letta", 2)}, left: 2, refuse: "remove " + imageRef("letta", 2)}, {name: "a confirmed collect", how: "collect", letGo: all}, {name: "a sweep", most: MostPerSweep, letGo: all}, {name: "an index waits", index: true, most: MostPerSweep, letGo: []string{archiveRef("letta", 2), mirror(1)}, skipped: 1}, {name: "bounded", most: 1, letGo: []string{archiveRef("letta", 2)}, left: 2}, // A kept build the log no longer holds: what it stood on is unknown, so no copy goes. - {name: "a kept build's bases unknown", most: MostPerSweep, letGo: []string{archiveRef("letta", 2), imageRef("letta", 2)}, unlog: 3}, + {name: "a kept build's bases unknown", most: MostPerSweep, letGo: []string{archiveRef("letta", 2), imageRef("letta", 2)}, unlog: 3, held: 8}, } { t.Run(c.name, func(t *testing.T) { h, b, door, acts := aMesh(t) door.indexes[imageRef("letta", 2)] = c.index + if c.held == 0 { + c.held = 7 + } + if c.seatForm { + for _, n := range []int{2, 3} { + key, _ := delivery.BuilderBuildKey("letta", fp(n)) + raw, _ := 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: catalogue.SeatReference("container-image-registry", "letta", "image", digestOf(imageRef("letta", n)))}, + {Name: "files", Kind: "archive", Digest: digestOf(archiveRef("letta", n)), Reference: catalogue.SeatReference("oci-artifact-registry", "letta", "files", digestOf(archiveRef("letta", n)))}}}) + _ = b.Put(BuildsRead, key, json.RawMessage(raw)) + } + } + if c.refuse != "" { + door.refuse = map[string]error{c.refuse: errors.New("the store is read only")} + } if c.unlog > 0 { b.log = append(b.log[:c.unlog-1:c.unlog-1], b.log[c.unlog:]...) } @@ -284,7 +341,7 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T) 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 { + if fmt.Sprint(a.LetGo) != fmt.Sprint(c.letGo) || a.Left != c.left || a.Skipped != c.skipped || a.Held != c.held || 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) } } @@ -298,8 +355,34 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T) held++ } } - if held != 6 { - t.Errorf("the store was asked %v: the six kept archives are held first", door.asked) + if held != c.held { + t.Errorf("the seats were asked %v: the %d kept archives and copies are held first", door.asked, c.held) + } + // Each removal asks the seat that holds it, by its name and digest, for retention, after a release. + for _, call := range door.calls { + if strings.Contains(call, ".remove ") && !strings.HasSuffix(call, " "+Retention) { + t.Errorf("a removal asked %q, not for retention", call) + } + } + if c.letGo[0] == archiveRef("letta", 2) && !slices.Contains(door.calls, "seat:oci-artifact-registry.remove letta/files@"+digestOf(archiveRef("letta", 2))+" "+Retention) { + t.Errorf("the seats were called %v", door.calls) + } + _, finding, _ := b.Get(FindingsState, sweepFindingKey) + if finding != (c.refuse != "") { + t.Errorf("the sweep's finding is written: %v, want %v", finding, c.refuse != "") + } + if c.refuse != "" { + door.refuse = nil + if _, err := h.Sweep(context.Background(), Sweeping{Most: MostPerSweep, By: "again"}); err != nil { + t.Fatal(err) + } + if _, still, _ := b.Get(FindingsState, sweepFindingKey); still { + t.Error("a clean sweep left the sweep's finding") + } + return + } + if c.seatForm && !strings.Contains(b.published[0], `"container-image-registry/letta/image@`+digestOf(imageRef("letta", 2))+`"`) { + t.Errorf("artifacts-gone does not name the build's own form: %v", b.published) } if len(*acts) != len(c.letGo) || !strings.HasPrefix((*acts)[0], "seat:mesh-hand-acts.hand-act ") || !strings.Contains((*acts)[0], c.letGo[0]) { 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 8eb92fba..1a01c2f9 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go @@ -27,7 +27,7 @@ func main() { UID: os.Getuid(), } h := &Holder{Bus: runtimeBus{}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View, - Now: time.Now, Logf: logf} + Seats: stdio.Ask, Now: time.Now, Logf: logf} h.Acts = handacts.Recorder{Ask: stdio.Ask} go follow(h) // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): artifact-store-tools. 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 ea6fdccd..75933be5 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 @@ -469,7 +469,10 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { } // ADR 0571 clause 2: it holds the seat mesh-artifact-sweep and serves its verbs, and calls no verb of the controller's; // it reads no controller state, the lease neither (issue 1126). - want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule} + // CAT-AS1: what it holds and removes it asks the two registry seats the artifact store's holder serves. + want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule, + "seat:oci-artifact-registry.hold", "seat:oci-artifact-registry.release", "seat:oci-artifact-registry.remove", + "seat:container-image-registry.hold", "seat:container-image-registry.release", "seat:container-image-registry.remove"} if !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes %v", m.Invokes) } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go b/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go new file mode 100644 index 00000000..20519cc7 --- /dev/null +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go @@ -0,0 +1,150 @@ +package main + +// **The sweep asks the registry seats** (CAT-AS1; novox/hq ADR 0617 §2, design 51): the artifact store's write side is +// its holder's, so the sweep no longer deletes or holds over the store's door. What it keeps from retention it asks the +// seat's `hold` for (a kept bundle or archive on oci-artifact-registry, a kept copy of an upstream base on +// container-image-registry), and what is eligible it asks the seat's `remove` for, with the sweep's reason; both verbs +// are in the manifest's `invokes`, and only this module is on their caller lists. A reference is read in both forms +// (ADR 0605 §9): the older `artifact-store://` and the seat's `//@`; the keep set +// keys each by the older form, so the two meet as one artifact. A call a seat refuses stops the sweep and writes the +// sweep's finding, sweep-refused; the next sweep with no refusal clears it. + +import ( + "encoding/json" + "fmt" + "strings" + + ociseat "git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/health" + + reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" +) + +// SweepRefused is the sweep's finding: a seat refused a hold or a remove the sweep asked. +const SweepRefused = "sweep-refused" + +// sweepFindingKey is the one key of the sweep's finding in the module's findings. +var sweepFindingKey = health.FindingKey(health.ScopeModule, Module, SweepRefused) + +// canonical is a kept reference in the older form, `artifact-store://`, whichever form it came in; another is as +// it was. +func canonical(reference string) string { + path, kept := catalogue.InArtifactStore(reference) + if !kept { + return reference + } + return catalogue.ArtifactStoreScheme + path +} + +// seatOf is the seat, name and digest a kept reference is held under: a bundle or an archive on oci-artifact-registry, +// an image or a copy of an upstream base on container-image-registry. +func seatOf(reference string) (seat, name, digest string, err error) { + path, kept := catalogue.InArtifactStore(reference) + if !kept { + return "", "", "", fmt.Errorf("%s is no reference into the mesh's registries", reference) + } + if name, digest, ok := strings.Cut(path, "/blobs/"); ok { + return ociseat.Seat, name, digest, nil + } + if name, digest, ok := strings.Cut(path, "@"); ok { + return reg.Seat, name, digest, nil + } + return "", "", "", fmt.Errorf("%s names no digest", reference) +} + +// isCopy is whether a kept reference is a copy of an upstream base. +func isCopy(reference string) bool { + path, _ := catalogue.InArtifactStore(reference) + return strings.HasPrefix(path, "upstream/") +} + +// askSeat asks verb of the seat reference is held on, with body built from its name and digest. +func (h *Holder) askSeat(reference, verb string, body func(name, digest string) any, answer any) error { + seat, name, digest, err := seatOf(reference) + if err != nil { + return err + } + if h.Seats == nil { + return fmt.Errorf("this process reaches no registry seat") + } + raw, err := h.Seats("seat:"+seat+"."+verb, body(name, digest)) + if err != nil { + return fmt.Errorf("%s's %s of %s@%s: %w", seat, verb, name, digest, err) + } + if answer != nil && len(raw) > 0 { + if err := json.Unmarshal(raw, answer); err != nil { + return fmt.Errorf("%s's answer to %s: %w", seat, verb, err) + } + } + return nil +} + +// holdOnSeat asks the seat to keep reference from retention, for why. +func (h *Holder) holdOnSeat(reference, why string) (bool, error) { + var a reg.HoldAnswer + err := h.askSeat(reference, reg.VerbHold, func(name, digest string) any { + return reg.HoldInput{Name: name, Digest: digest, Reason: why} + }, &a) + if err == nil && !a.Held { + err = fmt.Errorf("%s: the seat answered it is not held", reference) + } + return a.Held, err +} + +// Retention is the reason of a removal by digest, the sweep's alone (design 51 *Who may call which verb*): the holder +// removes an untagged thing only for it. The seat's definition names no such value yet (a request to mesh-contracts). +const Retention = "retention" + +// notHeld is the holder's refusal of a release of what it does not hold: nothing to let go. +const notHeld = "is not held" + +// removeOnSeat lets reference go: a hold an earlier sweep asked while it was kept is released first (the holder +// refuses to remove a held thing), then it is removed for retention. why goes with the release; the hand act records it. +func (h *Holder) removeOnSeat(reference, why string) error { + err := h.askSeat(reference, reg.VerbRelease, func(name, digest string) any { + return reg.ReleaseInput{Name: name, Digest: digest, Reason: why} + }, ®.ReleaseAnswer{}) + if err != nil && !strings.Contains(err.Error(), notHeld) { + return err + } + return h.askSeat(reference, reg.VerbRemove, func(name, digest string) any { + return reg.RemoveInput{Name: name, Digest: digest, Reason: Retention} + }, ®.RemoveAnswer{}) +} + +// sweepFinding writes the sweep's finding: what the seat refused, and why. +func (h *Holder) sweepFinding(what string, err error) { + at := h.now().UTC() + f := health.Finding{V: health.FindingVersion, Key: sweepFindingKey, Kind: SweepRefused, + Subject: health.Subject{Scope: health.ScopeModule, ID: Module}, + Severity: health.Warning, Resolver: health.ResolverOperator, + Summary: fmt.Sprintf("the artifact sweep stopped: a registry seat refused %s (the evidence names why)", what), + Headline: "The artifact sweep stopped", + Explanation: "The sweep keeps what the mesh still needs and removes the rest through the registry seats. A seat " + + "refused one of its asks, so it stopped before anything more was removed; what it kept stays kept.", + Needs: "read the seat holder's journal for the refusal; the next sweep asks again.", + Resolved: "Resolved: the artifact sweep ran through without a refusal", + Evidence: []health.Evidence{{At: at, Said: fmt.Sprintf("%s: %v", what, err)}}, + Source: "CAT-AS1", Observer: Module, At: at} + if werr := h.Bus.Put(FindingsState, sweepFindingKey, f); werr != nil { + h.logf("the sweep's finding (%s: %v) could not be written: %v", what, err, werr) + } +} + +// sweepClean clears the sweep's finding after a sweep no seat refused. +func (h *Holder) sweepClean() { + if _, found, err := h.Bus.Get(FindingsState, sweepFindingKey); err == nil && found { + if err := h.Bus.Delete(FindingsState, sweepFindingKey); err != nil { + h.logf("the sweep's finding could not be cleared: %v", err) + } + } +} + +// keptWhy is why a kept reference is held, as the seat's hold takes it. +func keptWhy(r *Ref) string { + if r == nil || len(r.Why) == 0 { + return "kept by the artifact sweep" + } + return "kept: " + strings.Join(r.Why, "; ") +} diff --git a/modules/artifact-store-tools/module.json b/modules/artifact-store-tools/module.json index 54cc5916..75d18186 100644 --- a/modules/artifact-store-tools/module.json +++ b/modules/artifact-store-tools/module.json @@ -20,7 +20,13 @@ "invokes": [ "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.modules", - "seat:mesh-catalog.module" + "seat:mesh-catalog.module", + "seat:oci-artifact-registry.hold", + "seat:oci-artifact-registry.release", + "seat:oci-artifact-registry.remove", + "seat:container-image-registry.hold", + "seat:container-image-registry.release", + "seat:container-image-registry.remove" ], "reads": [ "mesh-delivery-assigner.wants",