Registry milestone: mesh-delivery (issue 1966) #151

Merged
mesh-tools merged 359 commits from feat/registry-milestone into main 2026-11-07 14:19:29 +00:00
10 changed files with 173 additions and 36 deletions
Showing only changes of commit 71f4e8d128 - Show all commits
@@ -211,8 +211,9 @@ type bed struct {
instances jetstream.KeyValue
tagAsks jetstream.KeyValue
tagBounds jetstream.KeyValue
out *saidLog // the builder's process log, every instance's
stop func() // stops the builder started last
packages jetstream.KeyValue // the indexer's packages, served by its read tool and views family
out *saidLog // the builder's process log, every instance's
stop func() // stops the builder started last
fake *fakeBuildModules
srvURL string
dir string
@@ -293,6 +294,7 @@ func startBedWith(t *testing.T, closed bool) *bed {
for _, st := range []string{modulesState, manifestsState, candidatesState} {
viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, delivery.IndexerModule+".indexer_read_"+st, indexerViews, readers...)
}
b.packages = viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, delivery.PackagesState, delivery.IndexerModule+".indexer_read_packages", indexerViews, readers...)
viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", assignerViews, readers...)
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
b.instances = mk(instancesBucket, 1)
@@ -590,6 +592,12 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
t.Fatalf("the bound is not written again after its narration: %v", err)
}
}
// Design 32, ADR 0615 §1: a package's name the indexer writes later reaches the builder by its packages' views family
// (never their bucket), and looks at the module again; the done bundles are kept, not numbered twice.
if _, err := b.packages.Put(b.ctx, "netcheck.0.1", []byte(`{"v":1,"name":"netcheck","package":"netcheck/lib","series":"0.1"}`)); err != nil {
t.Fatal(err)
}
b.awaitSaid(t, "names a package of netcheck")
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 1 || got[0] != "0.1.5" {
t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.5 once", got)
}
@@ -64,7 +64,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
// The indexer's modules, manifests and candidates and the assigner's placement are their owners' states, read by the
// The indexer's modules, manifests, candidates and packages and the assigner's placement are their owners' states, read by the
// first view of their read tools and then their views families, never their buckets (ADR 0615 §1, DLV-M1). An owner that
// is down never holds the start: each is read on a goroutine once its first view is answered, and until then the
// unanswered first view is the builder's finding.
@@ -89,6 +89,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
// The indexer's packages, whose names a package is numbered under (design 32).
packages, err := reader.Follow(delivery.IndexerModule, delivery.PackagesState, delivery.IndexerModule+".indexer_read_packages", indexerViews)
if err != nil {
return err
}
placement, err := reader.Follow(delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", assignerViews)
if err != nil {
return err
@@ -145,9 +150,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
floorWaits.Keep(kvKeeper{ctx: ctx, kv: kv}, heldFloorWaits(ctx, kv, logger.Printf))
})
// A package is numbered under its name in its ecosystem, read from the indexer's packages state (design 32).
packagesKV := process.AwaitBucket(ctx, js, conn, delivery.PackagesBucket, "no package is numbered: each is built unnumbered and said", logger.Printf)
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
Names: versions.NamesIn(indexerPackages{packagesKV}),
Names: versions.NamesIn(indexerPackages{packages}),
Said: func(nb delivery.VersionNumbered) {
if err := emitNumbered(context.WithoutCancel(ctx), js, cred, nb); err != nil {
logger.Printf("%v", err)
@@ -328,6 +332,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf))
// The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests.
go func() { errs <- buildFromModules(ctx, b, modules, manifests, candidates) }()
// A package's name heard in the indexer's packages looks at its module again, so a done build is numbered then.
go func() { errs <- followPackages(ctx, packages, b, logger.Printf) }()
// A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as
// every other input. The builder asks again by itself without it.
// A state not on the bus yet at this start is watched once it appears (issue 1790).
@@ -53,19 +53,14 @@ func (s awaitedVersions) Get(ctx context.Context, key string) ([]byte, uint64, b
return e.Value(), e.Revision(), true, nil
}
// indexerPackages is the indexer's packages state (delivery.PackagesBucket, the manifest's reads grant) as the builder
// reads a package's name in its ecosystem from it (design 32): bound once the bus holds it; until then a package is
// built unnumbered and said (versions.Numberer.Names).
type indexerPackages struct{ b *process.AwaitedBucket }
var errPackagesMissing = errors.New("the indexer's state " + delivery.PackagesBucket + " is not on the bus yet")
// indexerPackages is the indexer's packages state as the builder reads a package's name in its ecosystem from it (design
// 32): the picture of its read tool's first view and its views family (ADR 0615 §1), never its bucket. Until the first
// view is answered a package is built unnumbered and said (versions.Numberer.Names), and numbered once its name is heard
// (followPackages).
type indexerPackages struct{ kv jetstream.KeyValue }
func (s indexerPackages) Keys(ctx context.Context) ([]string, error) {
kv := s.b.KV()
if kv == nil {
return nil, errPackagesMissing
}
keys, err := kv.Keys(ctx)
keys, err := s.kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
@@ -73,11 +68,7 @@ func (s indexerPackages) Keys(ctx context.Context) ([]string, error) {
}
func (s indexerPackages) Get(ctx context.Context, key string) ([]byte, bool, error) {
kv := s.b.KV()
if kv == nil {
return nil, false, errPackagesMissing
}
e, err := kv.Get(ctx, key)
e, err := s.kv.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrKeyDeleted) {
return nil, false, nil
}
@@ -87,6 +78,30 @@ func (s indexerPackages) Get(ctx context.Context, key string) ([]byte, bool, err
return e.Value(), true, nil
}
// followPackages watches the picture of the indexer's packages state: each series key heard that names a package's name
// (versions.NamedModule), its first view's included, looks at its module again (build.Builder.PackageNamed), so a done
// build that left the package unnumbered is numbered then with its digests. Woken by the views family, never a timer; a
// key heard again naming the name and module it named (the indexer writes a series at each publish) looks at nothing.
func followPackages(ctx context.Context, kv jetstream.KeyValue, b *build.Builder, logf func(string, ...any)) error {
w, err := kv.WatchAll(ctx)
if err != nil {
return err
}
heard := map[string]string{}
return follow(ctx, w, func(e jetstream.KeyValueEntry) {
if e.Operation() != jetstream.KeyValuePut {
delete(heard, e.Key())
return
}
module, name, named := versions.NamedModule(e.Key(), e.Value())
if named && heard[e.Key()] != module+" "+name {
heard[e.Key()] = module + " " + name
logf("%s.%s names a package of %s: its build is looked at again", delivery.PackagesBucket, e.Key(), module)
b.PackageNamed(ctx, module)
}
})
}
func versionsConflict(err error) error {
if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
return versions.ErrConflict
@@ -14,8 +14,9 @@ import (
// Numbers numbers a trunk build at its claim (novox/hq design 51 *The versions*, ADR 0606, ADR 0617 §1) and records the
// digests it was built with. Nil numbers nothing.
type Numbers interface {
// tag is a check's build's tag (TagOf), empty for a trunk build.
Number(ctx context.Context, module string, m delivery.Module, plan Plan, tag string) (versions.Numbering, error)
// tag is a check's build's tag (TagOf), empty for a trunk build; done the digests of its key when it is done already
// (versions.BuildClaim.Done), nil otherwise.
Number(ctx context.Context, module string, m delivery.Module, plan Plan, tag string, done map[string]string) (versions.Numbering, error)
Built(ctx context.Context, module string, m delivery.Module, plan Plan, got versions.Numbering, made []delivery.BuiltArtifact) error
}
@@ -41,8 +42,9 @@ func ClaimOf(module string, m delivery.Module, plan Plan) versions.BuildClaim {
return c
}
func (v VersionNumbers) Number(ctx context.Context, module string, m delivery.Module, plan Plan, tag string) (versions.Numbering, error) {
func (v VersionNumbers) Number(ctx context.Context, module string, m delivery.Module, plan Plan, tag string, done map[string]string) (versions.Numbering, error) {
c := ClaimOf(module, m, plan)
c.Done = done
if tag != "" {
// A check's build of a head a tag ask names: its packages tagged, below the next version (design 51 *Tagged versions*).
c.Tag, c.Head, c.Revision, c.Merge = tag, m.Commit, "", 0
@@ -73,7 +75,11 @@ func (b *Builder) number(ctx context.Context, module string, m delivery.Module,
if b.Numbers == nil {
return versions.Numbering{}, true
}
got, err := b.Numbers.Number(ctx, module, m, plan, tag)
var done map[string]string
if tag == "" {
done = b.doneDigests(ctx, module, m.FP2)
}
got, err := b.Numbers.Number(ctx, module, m, plan, tag, done)
for _, s := range got.Said {
b.logf("%s at %s: %s", module, m.FP2, s.Line)
// And in the build log (design 51 *The versions*: it says a version was kept, and why), between the build's start and end.
@@ -93,3 +99,40 @@ func (b *Builder) number(ctx context.Context, module string, m delivery.Module,
}
return got, true
}
// doneDigests are the digests of a trunk build's key by artifact when the key is done, nil otherwise or when it cannot be
// read: a look at a done build numbers what it left unnumbered with them (versions.BuildClaim.Done).
func (b *Builder) doneDigests(ctx context.Context, module, fp string) map[string]string {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil || b.Claimer.Store == nil {
return nil
}
raw, _, found, err := b.Claimer.Store.Get(ctx, key)
if err != nil || !found {
return nil
}
cur, err := delivery.DecodeBuilderBuild(raw)
if err != nil || cur.State != delivery.BuildDone {
return nil
}
out := map[string]string{}
for _, a := range cur.Artifacts {
if a.Digest != "" {
out[a.Name] = a.Digest
}
}
return out
}
// PackageNamed looks at a module again when the indexer's packages state names a name for one of its packages
// (versions.NamedModule): a done trunk build that left the package unnumbered, its name not known at its claim, is
// numbered then with the done build's digests, and version.numbered says them for its build module's late publish.
func (b *Builder) PackageNamed(ctx context.Context, module string) {
b.init()
b.mu.Lock()
m, ok := b.known[module]
b.mu.Unlock()
if ok {
b.consider(ctx, module, m)
}
}
@@ -474,6 +474,23 @@ func TestATrunkImageCallCarriesPublish(t *testing.T) { // 1705b; design 51 *The
if nums.built["img"] != digest {
t.Errorf("the built digest recorded = %v, want %s for img", nums.built, digest)
}
if nums.done != nil {
t.Errorf("a build still to make was numbered with done digests %v", nums.done)
}
// The indexer's packages state names a package of the module (PackageNamed): the done build is looked at again and
// numbered with its key's digests, so version.numbered carries them for the late publish; nothing is built again.
looks := nums.numbered
r.b.PackageNamed(context.Background(), "m")
r.b.Wait()
if nums.numbered != looks+1 || nums.done["img"] != digest || len(r.bm.callsOf("build_image_build")) != 1 {
t.Errorf("a named package's done build: numbered %d more times with %v, %d image builds; want once with img %s and no build",
nums.numbered-looks, nums.done, len(r.bm.callsOf("build_image_build")), digest)
}
r.b.PackageNamed(context.Background(), "other")
r.b.Wait()
if nums.numbered != looks+1 {
t.Errorf("a module the builder does not know was looked at")
}
}
// fakeNumbers is Numbers that answers what it is set to and keeps the digests a done build recorded.
@@ -483,10 +500,12 @@ type fakeNumbers struct {
numbered int
built map[string]string
tags []string
done map[string]string // the done key's digests the last look numbered with
}
func (f *fakeNumbers) Number(_ context.Context, _ string, _ delivery.Module, _ Plan, tag string) (versions.Numbering, error) {
func (f *fakeNumbers) Number(_ context.Context, _ string, _ delivery.Module, _ Plan, tag string, done map[string]string) (versions.Numbering, error) {
f.tags = append(f.tags, tag)
f.done = done
f.numbered++
return versions.Numbering{Wait: f.wait, Versions: f.versions}, nil
}
@@ -72,6 +72,10 @@ type BuildClaim struct {
// Tag numbers a check's build of a head a tag ask names (design 51 *Tagged versions*): its packages only, each a
// tagged version kept with Repository for its branch's end; Head is the head, said in numbered. Empty for a trunk build.
Tag, Head string
// Done are the digests of the build's done key, by artifact, when the claim is a look at a build already done: what it
// left unnumbered (a package whose name was not known yet, a floor that was silent) is numbered with them, so
// version.numbered carries the digest its build module publishes late. Nil for a build still to make.
Done map[string]string
}
// Numbering is what the claim got.
@@ -179,7 +183,7 @@ func (n *Numberer) Number(ctx context.Context, c BuildClaim) (Numbering, error)
floor, why = n.askFloor(ctx, seat, name, c)
}
d, e, err := n.Versions.Number(ctx, Claim{Name: name, Kind: kind, Series: c.Series, Module: c.Module, Fingerprint: c.Fingerprint,
Commit: c.Commit, Revision: c.Revision, Seat: seat, Merge: c.Merge, Floor: floor, Tag: c.Tag, Repository: c.Repository})
Commit: c.Commit, Revision: c.Revision, Digest: c.Done[a.Artifact], Seat: seat, Merge: c.Merge, Floor: floor, Tag: c.Tag, Repository: c.Repository})
if err != nil {
return out, err
}
@@ -3,6 +3,7 @@ package versions
import (
"context"
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -45,3 +46,21 @@ func NamesIn(s PackagesState) PackageNames {
return "", false, nil
}
}
// NamedModule is the module a value of the indexer's packages state names a package's name for (design 32), and that
// name: a series key whose value names a name and the <module>/<artifact> it is built from. The builder looks at that module again when
// the value is heard, so a done build that left the package unnumbered (its name not known at its claim) is numbered then.
func NamedModule(key string, value []byte) (module, name string, ok bool) {
if !delivery.IsSeriesPackageKey(key) {
return "", "", false
}
v, err := delivery.DecodeSeriesPackage(value)
if err != nil || v.Name == "" {
return "", "", false
}
module, artifact, cut := strings.Cut(v.Package, "/")
if !cut || module == "" || artifact == "" {
return "", "", false
}
return module, v.Name, true
}
@@ -512,6 +512,27 @@ func TestTheBuildersVersions(t *testing.T) {
if tagged.Versions["git.novox.be/novox/mesh-sdk/go"] != "0.1.4-x.1" {
t.Errorf("a tag ask names %v, want the Go module path", tagged.Versions)
}
// The indexer writes the name of `new` later: its series key names the module, and the look at the done build
// numbers it with the done key's digest, said in version.numbered for the late publish; the rest are kept.
state["new.0.1"] = []byte(`{"v":1,"name":"new","package":"mesh-sdk/new","series":"0.1"}`)
for key, want := range map[string]string{"new.0.1": "mesh-sdk", "git=2enovox=2ebe/novox/mesh-sdk/go": "", "broken.0.1": ""} {
if module, name, ok := NamedModule(key, state[key]); module != want || ok != (want != "") || (ok && name != "new") {
t.Errorf("NamedModule(%s) = %q, %q, %v, want %q", key, module, name, ok, want)
}
}
said = nil
done := c
done.Done = map[string]string{"go": "sha256:aa", "new": "sha256:bb"}
late, err := n.Number(ctx, done)
if err != nil || late.Versions["new"] != "0.1.4" || late.Versions["go"] != "0.1.4" || len(late.Unnumbered) != 0 {
t.Fatalf("a look at the done build = %+v, %v; want new numbered and go kept", late, err)
}
if len(said) != 1 || said[0].Name != "new" || said[0].Digest != "sha256:bb" {
t.Errorf("version.numbered said %+v, want new once with the done build's digest", said)
}
if es, _ := n.Versions.Of(ctx, "new"); len(es) != 1 || es[0].Digest != "sha256:bb" {
t.Errorf("versions of new = %+v, want its done digest recorded", es)
}
if _, _, err := NamesIn(packagesMem(nil).failing())(ctx, "m", "a"); err == nil {
t.Error("a packages state that cannot be read is no error")
}
@@ -100,7 +100,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
"mesh-delivery-gate.source", "mesh-delivery-indexer.packages"}; !reflect.DeepEqual(sorted(m.Reads), want) {
"mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
}
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
@@ -110,15 +110,16 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// other tool; it holds no capacity.
// 1955/L7 (ADR 0571 §2): the seat mesh-artifact-sweep's held, the controller's `held` is no longer asked (issue 1955).
// DLV-M1 (ADR 0615 §1): the indexer's modules, manifests and candidates and the assigner's placement are asked by their
// read tools and followed by their views families, never read.
// read tools and followed by their views families, never read; so are the indexer's packages, whose names a package is
// numbered under (design 32).
for _, r := range m.Reads {
if (strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.")) && r != "mesh-delivery-indexer.packages" {
if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") {
t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r)
}
}
// Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements.
if want := []string{"mesh-delivery-assigner.assigner_read_placement", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
@@ -134,7 +135,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.placement", "mesh-delivery-checker.tag-wanted", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery-indexer.manifest-accepted", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
"node-engine.manifest-reads-changed",
"mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed",
"npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed",
"oci-artifact-registry.published", "oci-artifact-registry.refused", "oci-artifact-registry.removed"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
+5 -4
View File
@@ -50,7 +50,8 @@
"mesh-delivery-assigner.views.placement",
"mesh-delivery-indexer.views.candidates",
"mesh-delivery-indexer.views.manifests",
"mesh-delivery-indexer.views.modules"
"mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.packages"
],
"invokes": [
"mesh-delivery-forge-bridge.bridge_tags",
@@ -69,7 +70,8 @@
"mesh-delivery-assigner.assigner_read_placement",
"mesh-delivery-indexer.indexer_read_candidates",
"mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules"
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_read_packages"
],
"state": [
{ "name": "builds", "history": 5 },
@@ -95,8 +97,7 @@
"mesh-build-image.steps",
"mesh-build-node.steps",
"mesh-build-node.toolchains",
"mesh-delivery-gate.source",
"mesh-delivery-indexer.packages"
"mesh-delivery-gate.source"
],
"uses": [
"node-build-go",