indexer: the build-edges state for builds of another watched branch, pruned on registration-changed, outdated kind 1 against the branch's newest build

BuildEdges.fp is the entry's fingerprint in the branch's revision whose source the build names; the branch's newest build
is kept as heard in heard as branch-build.<module>.<branch>.

Builds: design 47 G §374, §375, A-83, A-80, A-85
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:24:41 +01:00
parent 13550328a6
commit 31792235e6
10 changed files with 356 additions and 4 deletions
@@ -450,6 +450,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
SayFill: sayFill(conn, cred),
// The catalog's released versions (design 47 G §349): the state versions, bound when it appears, and the event version-released.
Versions: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)},
BuildEdges: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.BuildEdgesBucket, "no build of another watched branch keeps its edges", logger.Printf)},
Inputs: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)},
SayVersionReleased: sayVersionReleased(conn, cred), SayVersionsFilled: sayVersionsFilled(conn, cred),
// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
@@ -462,6 +463,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.RegistrationsBucket, "no registration is kept or written", logger.Printf)},
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)},
routerRecordUnread, ix, logger.Printf)
registry.Pruned = ix.PruneBranchBuilds
ix.Registry = registry
// The first fill of the versions (design 47 G §372): at the start, and again at each catalog revision kept until it is made.
fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKept), logf: logger.Printf}
@@ -0,0 +1,243 @@
package index
import (
"cmp"
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// The builds of a watched branch other than the first release no version (design 47 G §371); the indexer keeps each one's
// edge entries in the state build-edges, under <module>.<branch>, and what the branch's newest build of the module
// published as heard, branch-build.<module>.<branch> in the state heard (§374, A-83).
const (
BuildEdgesState = "build-edges"
BuildEdgesBucket = Module + "_" + BuildEdgesState
// BranchBuildPrefix opens branch-build.<module>.<branch>, each part escaped as a name (A-80).
BranchBuildPrefix = "branch-build."
)
// BranchBuild is what a branch-build key holds: the newest build of a module on a watched branch that was heard published,
// its source revision, the fingerprint of the branch's catalog entry at that revision and each artifact's digest.
type BranchBuild struct {
V int64 `json:"v"`
Version string `json:"version"`
Source int64 `json:"source"`
Fp string `json:"fp"`
Artifacts map[string]string `json:"artifacts"`
At time.Time `json:"at"`
}
// BranchBuildKey is the key of the state heard of a module's newest build on a branch.
func BranchBuildKey(module, branch string) string {
return BranchBuildPrefix + keys.EscapeName(module) + "." + keys.EscapeName(branch)
}
// BuildEdgesKey is the key of the state build-edges of a module's build on a branch, each part escaped as a name.
func BuildEdgesKey(module, branch string) string {
return meshcatalog.BuildEdgesByModuleBranchKey(keys.EscapeName(module), keys.EscapeName(branch))
}
// branchEntryAt is the module's entry in the newest revision of the branch taken in from that source revision (A-83: the
// revision whose source the build names, which the indexer holds itself).
func (ix *Indexer) branchEntryAt(ctx context.Context, module, branch string, source int64) (meshcatalog.CatalogEntry, error) {
if ix.Revisions == nil {
return meshcatalog.CatalogEntry{}, errors.New("the catalog graph is not kept here: no entry says the build's fingerprint")
}
names, err := ix.Revisions.Names(ctx)
if err != nil {
return meshcatalog.CatalogEntry{}, err
}
number := TrunkRevisionOf
if !isTrunkBranch(branch) {
number = func(n string) (int64, bool) { return BranchRevisionOf(branch, n) }
}
var numbers []int64
for _, n := range names {
if r, ok := number(n); ok {
numbers = append(numbers, r)
}
}
slices.SortFunc(numbers, func(a, b int64) int { return cmp.Compare(b, a) })
for _, n := range numbers {
name := RevisionObject(meshcatalog.CatalogRevision{Branch: branch, Revision: n})
rev, entries, err := ReadRevision(ctx, ix.Revisions, name)
if err != nil {
return meshcatalog.CatalogEntry{}, err
}
if rev.Source != source {
continue
}
for _, e := range entries {
if e.Module == module {
return e, nil
}
}
}
return meshcatalog.CatalogEntry{}, fmt.Errorf("the indexer keeps no revision of the branch %s at source revision %d with an entry of %s", branch, source, module)
}
// hearBranchBuild takes the package-published of a build of a watched branch other than the first (A-83): the build is kept as
// heard, and its edge entries are kept in build-edges, keyed by the fingerprint of the entry in the revision of that branch the
// build's source names. A build of an older source revision than the one kept leaves what is kept.
func (ix *Indexer) hearBranchBuild(ctx context.Context, p meshbuilds.PackagePublished) error {
if ix.BuildEdges == nil {
return errors.New("the state build-edges is not kept here")
}
entry, err := ix.branchEntryAt(ctx, p.Module, p.Branch, p.Source)
if err != nil {
return err
}
if entry.Fp == "" {
ix.logf("build-edges: the entry of %s on %s at source revision %d has no fingerprint, so its build keeps none", p.Module, p.Branch, p.Source)
return nil
}
at := ix.now().UTC()
err = write(ctx, ix.Heard, BranchBuildKey(p.Module, p.Branch), func(raw []byte, found bool) ([]byte, bool, error) {
var b BranchBuild
if found && json.Unmarshal(raw, &b) == nil && b.Source > p.Source {
return nil, false, nil
}
if !found || b.Source != p.Source || b.Fp != entry.Fp {
b = BranchBuild{Artifacts: map[string]string{}}
}
b.V, b.Version, b.Source, b.Fp, b.At = 1, p.Version, p.Source, entry.Fp, at
b.Artifacts[p.Artifact] = p.Digest
data, err := json.Marshal(b)
return data, false, err
})
if err != nil {
return err
}
edges := releaseEdges(entry, p.Dependencies)
return write(ctx, ix.BuildEdges, BuildEdgesKey(p.Module, p.Branch), func(raw []byte, found bool) ([]byte, bool, error) {
be := meshcatalog.BuildEdges{V: 1, Fp: entry.Fp, Source: p.Source}
if found {
kept, err := meshcatalog.DecodeBuildEdgesByModuleBranchValue(raw)
if err != nil {
return nil, false, err
}
if kept.Source > p.Source {
return nil, false, nil
}
if kept.Source == p.Source && kept.Fp == entry.Fp {
be.Edges = kept.Edges
}
}
for _, e := range edges {
if !slices.ContainsFunc(be.Edges, func(x meshcatalog.VersionEdge) bool { return x.Kind == e.Kind && x.Part == e.Part && x.Word == e.Word }) {
be.Edges = append(be.Edges, e)
}
}
data, err := json.Marshal(be)
return data, false, err
})
}
// PruneBranchBuilds forgets a module's builds of the branches its registration no longer watches beside its first
// (registration-changed, §374): its build-edges keys and its branch-build keys of every other branch.
func (ix *Indexer) PruneBranchBuilds(ctx context.Context, module string, watched []string) error {
keep := map[string]bool{}
if len(watched) > 1 {
for _, b := range watched[1:] {
keep[keys.EscapeName(b)] = true
}
}
var errs []error
forget := func(st Store, key string) {
if _, rev, found, err := st.Get(ctx, key); err != nil {
errs = append(errs, err)
} else if found {
if err := st.Delete(ctx, key, rev); err != nil && !errors.Is(err, ErrConflict) {
errs = append(errs, err)
}
}
}
if ix.BuildEdges != nil {
all, err := ix.BuildEdges.Keys(ctx)
if err != nil {
return err
}
for _, k := range all {
if m, b, ok := meshcatalog.ParseBuildEdgesByModuleBranchKey(k); ok && m == keys.EscapeName(module) && !keep[b] {
forget(ix.BuildEdges, k)
}
}
}
if ix.Heard != nil {
all, err := ix.Heard.Keys(ctx)
if err != nil {
return err
}
prefix := BranchBuildPrefix + keys.EscapeName(module) + "."
for _, k := range all {
if b, ok := strings.CutPrefix(k, prefix); ok && !keep[b] {
forget(ix.Heard, k)
}
}
}
return errors.Join(errs...)
}
// outdatedBranchBuilds are the behind entries of kind 1 of the builds of other watched branches (A-83): the digest a build
// recorded for a package against the digest in the newest build of that package's module on the same branch, as heard, never a
// released digest.
func (ix *Indexer) outdatedBranchBuilds(ctx context.Context, in meshcatalog.OutdatedInput) ([]meshcatalog.Outdated, error) {
if ix.BuildEdges == nil || ix.Heard == nil {
return nil, nil
}
all, err := ix.BuildEdges.Keys(ctx)
if err != nil {
return nil, err
}
slices.Sort(all)
var out []meshcatalog.Outdated
for _, k := range all {
em, eb, ok := meshcatalog.ParseBuildEdgesByModuleBranchKey(k)
if !ok {
continue
}
module, err1 := keys.UnescapeName(em)
branch, err2 := keys.UnescapeName(eb)
if err1 != nil || err2 != nil || (in.Branch != "" && in.Branch != eb) {
continue
}
raw, _, found, err := ix.BuildEdges.Get(ctx, k)
if err != nil || !found {
continue
}
be, err := meshcatalog.DecodeBuildEdgesByModuleBranchValue(raw)
if err != nil {
return nil, fmt.Errorf("the key %s of the state %s: %w", k, BuildEdgesState, err)
}
for _, e := range be.Edges {
if e.Kind != 1 || (in.Dependency != "" && in.Dependency != e.Part) {
continue
}
var dep BranchBuild
raw, _, found, err := ix.Heard.Get(ctx, BranchBuildKey(strings.SplitN(e.Part, ":", 2)[0], branch))
if err != nil || !found || json.Unmarshal(raw, &dep) != nil {
continue
}
digest := dep.Artifacts[e.Part]
if digest == "" || digest == e.Digest {
continue
}
o := meshcatalog.Outdated{Module: module, Build: be.Fp, Kind: 1, Recorded: e.Version + " " + e.Digest, Newest: dep.Version + " " + digest,
Dependency: meshcatalog.OutdatedDependency{Artifact: e.Part, Branch: branch}}
if e.Stale {
o.Act = "rebuild"
}
out = append(out, o)
}
}
return out, nil
}
@@ -0,0 +1,90 @@
package index
import (
"context"
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// Design 47 G §374, A-83, A-85: a build of a watched branch other than the first releases no version; its edge entries are kept
// in build-edges under <module>.<branch>, the fingerprint being the entry's in the branch's revision whose source the build
// names; outdated compares kind 1 with the same branch's newest build, as heard; registration-changed prunes the branch's
// keys when it is no longer watched.
func TestABuildOfAnotherWatchedBranchKeepsItsEdgesAndIsOutdatedAgainstThatBranch(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), Heard: newMem(), BuildEdges: newMem(), Revisions: newMemObjects(1 << 20), Registry: &Registry{Registrations: MemoryStore()}}
for _, m := range []string{"lib", "app"} {
raw, _ := json.Marshal(meshcatalog.Registration{Module: m, Branches: []string{"main", "beta"}})
if err := ix.Registry.Registrations.Create(ctx, RegistrationKey(m), raw); err != nil {
t.Fatal(err)
}
}
fp := func(c string) string { return "src2:" + strings.Repeat(c, 64) }
keep := func(source int64, libFp, appFp string) {
t.Helper()
err := ix.KeepBranchEntries(ctx, "beta", source, map[string]meshcatalog.CatalogEntry{
"lib": {Module: "lib", Fp: libFp, Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "lib:pkg"}}},
"app": {Module: "app", Fp: appFp, Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "app:program"}},
Edges: []meshcatalog.Edge{{Kind: 1, From: "lib:pkg", To: "app:program", Part: "lib:pkg", Stale: true, Word: "imports"}}}})
if err != nil {
t.Fatal(err)
}
}
keep(7, fp("a"), fp("b"))
keep(8, fp("c"), fp("b"))
pub := func(module, artifact, digest string, source int64, deps ...meshbuilds.Dependency) {
t.Helper()
if err := ix.HearPackagePublished(ctx, meshbuilds.PackagePublished{Module: module, Artifact: artifact, Digest: digest, Version: "0.1.1", Source: source, Branch: "beta", Dependencies: deps}); err != nil {
t.Fatal(err)
}
}
pub("lib", "lib:pkg", "d1", 7)
pub("app", "app:program", "a1", 7, meshbuilds.Dependency{Kind: 1, Part: "lib:pkg", Version: "0.1.1", Digest: "d1"})
if v, _ := ModuleVersions(ctx, ix.Versions, "app"); len(v) != 0 {
t.Errorf("a build of beta released %v", v)
}
raw, _, found, _ := ix.BuildEdges.Get(ctx, BuildEdgesKey("app", "beta"))
be, err := meshcatalog.DecodeBuildEdgesByModuleBranchValue(raw)
if !found || err != nil || be.Fp != fp("b") || be.Source != 7 || len(be.Edges) != 1 || be.Edges[0].Digest != "d1" {
t.Fatalf("build-edges of app on beta: %+v (found %v, %v), want the fingerprint of the entry at source 7 and its kind 1 edge", be, found, err)
}
behind := func() []meshcatalog.Outdated {
t.Helper()
a, err := ix.Outdated(ctx, meshcatalog.OutdatedInput{})
if err != nil {
t.Fatal(err)
}
return a.Behind
}
if b := behind(); len(b) != 0 {
t.Fatalf("behind before lib's next build: %+v", b)
}
pub("lib", "lib:pkg", "d2", 8)
b := behind()
if len(b) != 1 || b[0].Module != "app" || b[0].Kind != 1 || b[0].Build != fp("b") || b[0].Version != "" || b[0].Dependency.Branch != "beta" ||
!strings.Contains(b[0].Recorded, "d1") || !strings.Contains(b[0].Newest, "d2") || b[0].Act != "rebuild" {
t.Fatalf("behind after lib's newer build on beta: %+v", b)
}
if err := ix.PruneBranchBuilds(ctx, "app", []string{"main", "beta"}); err != nil {
t.Fatal(err)
}
if _, _, found, _ := ix.BuildEdges.Get(ctx, BuildEdgesKey("app", "beta")); !found {
t.Errorf("a branch still watched lost its build-edges")
}
if err := ix.PruneBranchBuilds(ctx, "app", []string{"main"}); err != nil {
t.Fatal(err)
}
if _, _, found, _ := ix.BuildEdges.Get(ctx, BuildEdgesKey("app", "beta")); found {
t.Errorf("build-edges of app on beta stayed after beta was unwatched")
}
if _, _, found, _ := ix.Heard.Get(ctx, BranchBuildKey("app", "beta")); found {
t.Errorf("branch-build of app on beta stayed after beta was unwatched")
}
if _, _, found, _ := ix.Heard.Get(ctx, BranchBuildKey("lib", "beta")); !found {
t.Errorf("lib's branch-build went with app's")
}
}
@@ -109,6 +109,9 @@ type Indexer struct {
SayFill func(ctx context.Context, event, id string, f meshcatalog.Fill) error
// Versions is the state versions (versions.go, design 47 G §349): each module's released versions; nil keeps none.
Versions Store
// BuildEdges is the state build-edges (branch_builds.go, design 47 G §374): the edge entries of the newest build of a module on a
// watched branch other than the first; nil keeps none.
BuildEdges Store
// SayVersionReleased says the seat mesh-catalog's event version-released under the duplicate id given (§349); nil says nothing.
SayVersionReleased func(ctx context.Context, e meshcatalog.VersionReleasedEvent, id string) error
// SayVersionsFilled says the seat mesh-catalog's event versions-filled, once, under the duplicate id given (§372); nil says nothing.
@@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn",
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs"} {
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -121,6 +121,9 @@ type Registry struct {
HistoryOf func(ctx context.Context, module string) ([]meshcatalog.Registration, error)
// Say says the seat mesh-catalog's event registration-changed.
Say func(ctx context.Context, e meshcatalog.RegistrationChangedEvent) error
// Pruned is told after a registration changed which branches the module watches, so what is kept of its builds of a branch
// it no longer watches goes (registration-changed, design 47 G §374); nil prunes none.
Pruned func(ctx context.Context, module string, watched []string) error
// Branch reads a branch through the seat git; an empty branch asks the repository's default branch.
Branch func(repository, branch string) (gitseat.BranchAnswer, error)
// Ask publishes an ask under the indexer's own name, Cancel takes it back, Where reads the router's own record of an
@@ -796,6 +799,11 @@ func (r *Registry) write(ctx context.Context, key string, p Proposal, next meshc
r.logf("registration-changed of %s %d is not said: %v", next.Module, next.Version, err)
}
}
if r.Pruned != nil {
if err := r.Pruned(ctx, next.Module, next.Branches); err != nil {
r.logf("the builds of %s on the branches it no longer watches are not pruned: %v", next.Module, err)
}
}
return r.endKey(ctx, key, p, ProposalApplied, fmt.Sprintf("applied as version %d of %s on %s", next.Version, next.Module, occ.Says(w)))
}
@@ -100,7 +100,7 @@ func ModuleVersions(ctx context.Context, st Store, module string) ([]meshcatalog
// HearPackagePublished takes one artifact's package-published (design 47 G §349): the artifact is kept as published at its
// version, and once every artifact the module's catalog entry names is published at it the version is released: recorded,
// and said as version-released unless a higher one is released already (then it is recorded marked late and not said). A
// build of a watched branch other than the registration's first releases none (§371).
// build of a watched branch other than the registration's first releases none (§371) and keeps its edges in build-edges (§374, A-83).
func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.PackagePublished) error {
if ix.Versions == nil || ix.Heard == nil {
return errors.New("the state versions is not kept here")
@@ -110,7 +110,7 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag
}
if ix.Registry != nil {
if r, found, err := ix.Registry.Registration(ctx, p.Module); err == nil && found && len(r.Branches) > 0 && r.Branches[0] != p.Branch {
return nil
return ix.hearBranchBuild(ctx, p) // releases no version; its edges are kept in build-edges (A-83)
}
}
if _, _, found, err := ix.Versions.Get(ctx, meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)); err != nil || found {
@@ -271,6 +271,11 @@ func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (
out.Behind = append(out.Behind, o)
}
}
behind, err := ix.outdatedBranchBuilds(ctx, in)
if err != nil {
return out, err
}
out.Behind = append(out.Behind, behind...)
return out, nil
}
@@ -125,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
{Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5},
{Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1, MaxMiB: 8}, // design 47 G §695
{Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, History: 1, MaxMiB: 4},
{Name: index.VersionsState, History: 1, MaxMiB: 64}, {Name: index.InputsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
{Name: index.VersionsState, History: 1, MaxMiB: 64}, {Name: index.BuildEdgesState, History: 1}, {Name: index.InputsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
t.Errorf("state = %+v, want %+v", m.State, want)
}
// Milestone 47/G W4.2 (design 47, Two graphs; ADR 0619 §2): the catalog graph revisions are objects in the old section,
@@ -119,6 +119,7 @@
{ "name": "registrations", "history": 1 },
{ "name": "proposals", "history": 1, "max-mib": 4 },
{ "name": "versions", "history": 1, "max-mib": 64 },
{ "name": "build-edges", "history": 1 },
{ "name": "inputs", "history": 1 }
],
"streams": [