artifact-store-tools reads the builder's builds through its published package, the wants, haves and manifests through the seats' types and clients; the sweep seat's two bodies are the holder's own (B5f; ADR 0634 §1, §4; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 15:46:23 +01:00
parent 42387b660c
commit d4abb82b07
7 changed files with 126 additions and 54 deletions
@@ -0,0 +1,50 @@
package main
import (
"encoding/json"
"fmt"
"sort"
"time"
)
// The holders whose states this module reads, one constant each (a reader names its holder once; ADR 0634 §4).
const (
assignerModule = "mesh-delivery-assigner"
indexerModule = "mesh-delivery-indexer"
)
// ArtifactsGoneEvent is the event the seat mesh-artifact-sweep says when the store does not hold an artifact a
// builder's key names. The seat is a seat-list seat, which has no generated types, so its two bodies are the holder's
// own: this module is the one that says the event and answers the verb held.
const ArtifactsGoneEvent = "artifacts-gone"
// HeldAnswer is the answer of the seat's verb held: the references the store holds, and those it answered 404 for. A
// reference the store could not be asked about is in neither, and named in Unknown with why.
type HeldAnswer struct {
Store string `json:"store,omitempty"`
Held []string `json:"held"`
Gone []string `json:"gone"`
Unknown map[string]string `json:"unknown,omitempty"`
}
// ArtifactsGone is the body of ArtifactsGoneEvent: the module whose builder's key names a gone artifact, the key and
// its fingerprint, and every reference of it the store does not hold.
type ArtifactsGone struct {
Event string `json:"event"`
Module string `json:"module"`
Key string `json:"key"`
Fingerprint string `json:"fingerprint"`
Gone []string `json:"gone"`
At time.Time `json:"at"`
}
// EncodeArtifactsGone is the event's body, the references sorted.
func EncodeArtifactsGone(e ArtifactsGone) ([]byte, error) {
e.Event = ArtifactsGoneEvent
e.Gone = append([]string(nil), e.Gone...)
sort.Strings(e.Gone)
if e.Module == "" || e.Fingerprint == "" || len(e.Gone) == 0 {
return nil, fmt.Errorf("%s names a module, a fingerprint and at least one gone reference", ArtifactsGoneEvent)
}
return json.Marshal(e)
}
@@ -18,6 +18,12 @@ import (
"encoding/json"
"errors"
"fmt"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
meshwants "git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"sort"
"strings"
"sync"
@@ -28,24 +34,23 @@ import (
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
// What the holder reads and keeps.
const (
WantsRead = delivery.AssignerModule + ".wants"
BuildsRead = delivery.BuilderModule + ".builds"
WantsRead = assignerModule + ".wants"
BuildsRead = builds.BuilderModule + ".builds"
HavesRead = "node-engine.have"
// ManifestsRead is the indexer's manifests: the images a manifest pins by itself.
ManifestsRead = delivery.IndexerModule + ".manifests"
ManifestsRead = indexerModule + "." + catalog.StateManifests
// FindingsState is the holder's own findings (meshconditions.StateFindings): image-missing.
FindingsState = meshconditions.StateFindings
// 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.ArtifactSweepSeat + ".event." + delivery.ArtifactsGoneEvent
GoneSubject = "mesh.seat." + catalogue.ArtifactSweepSeat + ".event." + ArtifactsGoneEvent
)
// The sweep's bounds, the controller's (cmd/mesh-controller/collect.go mostPerSweep, sweepBudget).
@@ -99,7 +104,7 @@ type Holder struct {
type Build struct {
Module, FP, Key string
At time.Time
Artifacts []delivery.BuiltArtifact
Artifacts []builds.BuiltArtifact
// Manifest is the digest of the manifest it built, in the indexer's `manifests`.
Manifest string
}
@@ -112,7 +117,7 @@ func (h *Holder) Builds() ([]Build, error) {
}
var out []Build
for _, k := range keys {
module, fp, err := delivery.ParseBuilderBuildKey(k)
module, fp, err := builds.ParseBuilderBuildKey(k)
if err != nil {
continue
}
@@ -123,8 +128,8 @@ func (h *Holder) Builds() ([]Build, error) {
if !found {
continue
}
b, err := delivery.DecodeBuilderBuild(raw)
if err != nil || b.State != delivery.BuildDone {
b, err := builds.DecodeBuilderBuild(raw)
if err != nil || b.State != builds.BuildDone {
continue
}
at, err := h.endOf(k)
@@ -149,7 +154,7 @@ func (h *Holder) endOf(key string) (time.Time, error) {
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) {
if b, err := builds.DecodeBuilderBuild(r.Value); err == nil && b.State == builds.BuildDone && r.At.After(at) {
at = r.At
}
}
@@ -181,11 +186,11 @@ func (h *Holder) Wants() ([]Want, error) {
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 == "" {
var w meshwants.WantsByWantNodeModuleValue
if !found || json.Unmarshal(raw, &w) != nil || w.Removed || w.Fp == "" {
continue
}
out = append(out, Want{Node: node, Module: module, FP: w.FP})
out = append(out, Want{Node: node, Module: module, FP: w.Fp})
}
return out, nil
}
@@ -352,17 +357,17 @@ func (h *Holder) judge(ctx context.Context) (*judged, error) {
// against is what each build the builder's log says ended stood on, by build key: the bases' references.
func (h *Holder) against() (map[string][]string, error) {
entries, err := h.Bus.LogSince(delivery.BuildLogRead, h.now().Add(-delivery.BuilderLogDays*24*time.Hour))
entries, err := h.Bus.LogSince(builds.BuildLogRead, h.now().Add(-builds.BuilderLogDays*24*time.Hour))
if err != nil {
return nil, err
}
out := map[string][]string{}
for _, raw := range entries {
e, err := delivery.DecodeBuildLogEntry(raw.Value)
if err != nil || e.Event != delivery.BuildEndedEvent {
e, err := builds.DecodeBuildLogEntry(raw.Value)
if err != nil || e.Event != builds.BuildEndedEvent {
continue
}
key, err := delivery.BuilderBuildKey(e.Module, e.FP)
key, err := builds.BuilderBuildKey(e.Module, e.FP)
if err != nil {
continue
}
@@ -613,7 +618,7 @@ func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error)
// 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{}
byKey := map[string]*ArtifactsGone{}
var keys []string
for _, r := range refs {
ref := j.refs[r]
@@ -623,7 +628,7 @@ func (h *Holder) sayGone(j *judged, refs []string, how string) {
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()}
g = &ArtifactsGone{Module: b.Module, Key: b.Key, Fingerprint: b.FP, At: h.now().UTC()}
byKey[b.Key] = g
keys = append(keys, b.Key)
}
@@ -632,7 +637,7 @@ func (h *Holder) sayGone(j *judged, refs []string, how string) {
}
sort.Strings(keys)
for _, k := range keys {
raw, err := delivery.EncodeArtifactsGone(*byKey[k])
raw, err := EncodeArtifactsGone(*byKey[k])
if err == nil {
err = h.Bus.Publish(GoneSubject, json.RawMessage(raw))
}
@@ -666,25 +671,25 @@ func (h *Holder) AfterBuild(key string) {
}
// 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) {
func (h *Holder) Held(refs []string) (HeldAnswer, error) {
if len(refs) == 0 {
return delivery.HeldAnswer{}, errors.New("held judges the references given, and none was")
return 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))
return 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)
return 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
return HeldAnswer{}, err
}
asking, stop := context.WithTimeout(context.Background(), 20*time.Second)
defer stop()
a := delivery.HeldAnswer{Store: door.Address(), Held: []string{}, Gone: []string{}}
a := HeldAnswer{Store: door.Address(), Held: []string{}, Gone: []string{}}
for _, r := range refs {
switch err := door.Holds(asking, canonical(r)); {
case err == nil:
@@ -725,7 +730,7 @@ func (h *Holder) ImagesOf(node string, address string) (Images, error) {
}
byImage := map[string]*Image{}
add := func(module, fp, in string) error {
key, err := delivery.BuilderBuildKey(module, fp)
key, err := builds.BuilderBuildKey(module, fp)
if err != nil {
return nil
}
@@ -733,7 +738,7 @@ func (h *Holder) ImagesOf(node string, address string) (Images, error) {
if err != nil || !found {
return err
}
b, err := delivery.DecodeBuilderBuild(raw)
b, err := builds.DecodeBuilderBuild(raw)
if err != nil {
return nil
}
@@ -774,7 +779,7 @@ func (h *Holder) ImagesOf(node string, address string) (Images, error) {
return out, fmt.Errorf("the haves: %w", err)
}
for _, k := range keys {
n, module, isNode, err := delivery.ParseHaveKey(k)
n, module, isNode, err := engineclient.ParseHaveKey(k)
if err != nil || isNode || n != node {
continue
}
@@ -782,12 +787,12 @@ func (h *Holder) ImagesOf(node string, address string) (Images, error) {
if err != nil {
return out, err
}
var have delivery.Have
if !found || json.Unmarshal(raw, &have) != nil || have.FP == "" {
var have nodeengine.HaveByNodeModuleValue
if !found || json.Unmarshal(raw, &have) != nil || have.Fp == "" {
continue
}
out.SentKnown = true
if err := add(module, have.FP, "sent"); err != nil {
if err := add(module, have.Fp, "sent"); err != nil {
return out, err
}
}
@@ -806,7 +811,7 @@ func (h *Holder) Pinned(manifest string) ([]string, error) {
if manifest == "" {
return nil, nil
}
key, err := delivery.ManifestsKey(manifest)
key, err := catalogclient.ManifestsKey(manifest)
if err != nil {
return nil, nil
}
@@ -814,14 +819,14 @@ func (h *Holder) Pinned(manifest string) ([]string, error) {
if err != nil || !found {
return nil, err
}
m, err := delivery.DecodeManifest(raw)
m, err := catalogclient.DecodeManifest(raw)
if err != nil {
return nil, nil
}
var parsed struct {
Resources []map[string]any `json:"resources"`
}
if json.Unmarshal([]byte(m.JSON), &parsed) != nil {
if json.Unmarshal([]byte(m.Manifest), &parsed) != nil {
return nil, nil
}
var out []string
@@ -5,6 +5,11 @@ import (
"encoding/json"
"errors"
"fmt"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
meshwants "git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants"
catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"slices"
"sort"
"strings"
@@ -16,7 +21,6 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
@@ -33,7 +37,7 @@ type memBus struct {
func (b *memBus) LogSince(read string, at time.Time) ([]stdio.LogEntry, error) {
var out []stdio.LogEntry
for _, e := range b.log {
if read == delivery.BuildLogRead && !e.At.Before(at) {
if read == builds.BuildLogRead && !e.At.Before(at) {
out = append(out, e)
}
}
@@ -175,8 +179,8 @@ func aMesh(t *testing.T) (*Holder, *memBus, *memDoor, *[]string) {
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{
key, _ := builds.BuilderBuildKey("letta", fp(n))
raw, err := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: strings.Repeat("d", 64), Artifacts: []builds.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)}}})
@@ -204,7 +208,7 @@ func aMesh(t *testing.T) (*Holder, *memBus, *memDoor, *[]string) {
door.holds[mirror(n)] = true
}
for n := 1; n <= 7; n++ {
e := delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: "letta", FP: fp(n), Repository: "novox/x", Commit: "c",
e := builds.BuildLogEntry{Event: builds.BuildEndedEvent, Module: "letta", FP: fp(n), Repository: "novox/x", Commit: "c",
Node: "ace", At: day.Add(time.Duration(n) * time.Hour), Outcome: "done"}
switch n {
case 2:
@@ -212,15 +216,15 @@ func aMesh(t *testing.T) (*Holder, *memBus, *memDoor, *[]string) {
case 7:
e.Against = []string{mirror(2)}
}
raw, _ := delivery.EncodeBuildLogEntry(e)
raw, _ := builds.EncodeBuildLogEntry(e)
b.log = append(b.log, stdio.LogEntry{Sequence: uint64(n), Key: "letta", Value: raw, At: e.At})
}
manifest, _ := delivery.EncodeManifest(delivery.Manifest{Module: "letta",
JSON: `{"module":"letta","resources":[{"id":"app","type":"container","image":"` + pinnedImage + `"},` +
manifest, _ := catalogclient.EncodeManifest(catalog.ManifestsByDigestValue{Module: "letta",
Manifest: `{"module":"letta","resources":[{"id":"app","type":"container","image":"` + pinnedImage + `"},` +
`{"id":"built","type":"container","image":"${artifact:image}"}]}`})
mk, _ := delivery.ManifestsKey(strings.Repeat("d", 64))
mk, _ := catalogclient.ManifestsKey(strings.Repeat("d", 64))
_ = b.Put(ManifestsRead, mk, json.RawMessage(manifest))
want, _ := json.Marshal(delivery.Want{V: delivery.Version, FP: fp(1), Self: true})
want, _ := json.Marshal(meshwants.WantsByWantNodeModuleValue{V: 1, 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 }, Seats: door.seat,
@@ -302,8 +306,8 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T)
}
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{
key, _ := builds.BuilderBuildKey("letta", fp(n))
raw, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: strings.Repeat("d", 64), Artifacts: []builds.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))
@@ -426,7 +430,7 @@ func TestAWantedImageTheStoreLacksIsAFindingAndImagesNamesWantsAndHaves(t *testi
t.Fatalf("the pinnedImage image missing is %s", raw)
}
have, _ := json.Marshal(delivery.Have{V: delivery.Version, FP: fp(7), Phase: "running"})
have, _ := json.Marshal(nodeengine.HaveByNodeModuleValue{V: 1, 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) != 3 || !images.SentKnown || images.Images[0].Image != "10.0.0.5:5000/letta/image@sha256:"+strings.Repeat("1", 64) ||
@@ -16,7 +16,6 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -26,8 +25,8 @@ func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
func main() {
s := &Store{Ask: stdio.Ask}
v := &views.Reader{Module: "artifact-store-tools", Logf: logf, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}}
v.Follow(delivery.AssignerModule, "wants", WantsTool)
v.Follow(delivery.IndexerModule, "manifests", ManifestsTool)
v.Follow(assignerModule, "wants", WantsTool)
v.Follow(indexerModule, "manifests", ManifestsTool)
h := &Holder{Bus: runtimeBus{views: v}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE"), s), Listed: s.View,
Seats: stdio.Ask, Now: time.Now, Logf: logf}
h.Acts = meshhandactsclient.Recorder{Ask: stdio.Ask}
@@ -42,6 +41,6 @@ func main() {
// The owners' read tools of the states the holder follows, which the manifest invokes (ADR 0615).
const (
WantsTool = delivery.AssignerModule + ".assigner_read_wants"
ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests"
WantsTool = assignerModule + ".assigner_read_wants"
ManifestsTool = indexerModule + ".indexer_read_manifests"
)
@@ -33,7 +33,8 @@ func TestEveryFileTakesItsTextFromATemplate(t *testing.T) {
t.Fatal(err)
}
var m struct {
Resources []map[string]any `json:"resources"`
Resources []map[string]any `json:"resources"`
Packages map[string]string `json:"packages"`
Build struct {
Artifacts []map[string]any `json:"artifacts"`
} `json:"build"`
@@ -41,6 +42,10 @@ func TestEveryFileTakesItsTextFromATemplate(t *testing.T) {
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
// The builder's keys are read through its published package, taken by its series (ADR 0634 §1).
if m.Packages["builds"] != "0.1" {
t.Errorf("packages %v: want builds of series 0.1", m.Packages)
}
for _, r := range m.Resources {
if _, inline := r["content"]; inline {
t.Errorf("resource %v keeps its text inline: keep it in files/ and name it as the template", r["id"])
+6
View File
@@ -3,6 +3,12 @@ module artifactstoretools
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
+3
View File
@@ -1,6 +1,9 @@
{
"module": "artifact-store-tools",
"version": "1",
"packages": {
"builds": "0.1"
},
"claims": [
{
"name": "mesh-artifact-sweep",