indexer: the release's edge entries, outdated for kinds 1, 2 and 4, toolchain-agreed and version-delivered kept as heard, the verb input and its state

Kind 3, kinds 5 and 6, the first fill's edge entries (§377) and the deletion of an input at its ask's step-ended are not built.

Builds: design 47 G §374, §375, §354, A-68

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:07:45 +01:00
parent e9d2997760
commit d9066aa1c6
12 changed files with 640 additions and 33 deletions
@@ -288,7 +288,9 @@ func main() {
tools = append(tools, catalogTools(trunk.State)...)
// The seat mesh-catalog's verb revision, a first view of the catalog half answered in this runtime of its own to exactly
// the parts whose entry declares it (novox/hq ADR 0663).
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}))
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}),
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, OwnBuildState: keptRead{stdio.State(index.OwnBuildStateName)}}),
index.InputTool(keptRead{stdio.State(index.InputsState)}), index.InputsReadTool(keptRead{stdio.State(index.InputsState)}))
tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error {
_, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"})
return err
@@ -24,6 +24,7 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -445,6 +446,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)},
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;
// until then every merge is taken in at once, as before.
@@ -679,6 +681,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := hub.Handle(meshbuilds.SubjectPackagePublished, func(m jetstream.Msg) { heardPackagePublished(ctx, ix, m, logger.Printf) }); err != nil {
return err
}
if err := hub.Handle(meshbuilds.SubjectToolchainAgreed, func(m jetstream.Msg) { heardToolchainAgreed(ctx, ix, m, logger.Printf) }); err != nil {
return err
}
if err := hub.Handle(meshwants.SubjectVersionDelivered, func(m jetstream.Msg) { heardVersionDelivered(ctx, ix, m, logger.Printf) }); err != nil {
return err
}
if err := hub.Handle(decidedSubject(cred.Module), func(m jetstream.Msg) { heardDecided(ctx, registry, m, logger.Printf) }); err != nil {
return err
}
@@ -12,6 +12,7 @@ import (
"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-delivery/contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-indexer/internal/index"
@@ -59,6 +60,40 @@ func heardPackagePublished(ctx context.Context, ix *index.Indexer, m jetstream.M
_ = m.Ack()
}
// heardToolchainAgreed keeps the newest toolchain-agreed of its language (A-68); one that does not decode is said and
// acknowledged, one that cannot be kept is left, so the bus gives it again.
func heardToolchainAgreed(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) {
var t meshbuilds.ToolchainAgreed
if err := json.Unmarshal(m.Data(), &t); err != nil {
logf("%s holds what is no toolchain-agreed, left: %v", m.Subject(), err)
_ = m.Ack()
return
}
if err := ix.HearToolchainAgreed(ctx, t); err != nil {
logf("versions: the toolchain-agreed of %s could not be kept, so the bus gives it again: %v", t.Language, err)
_ = m.Nak()
return
}
_ = m.Ack()
}
// heardVersionDelivered writes the delivered release's entries of kinds 2, 4 and 6 and keeps the newest delivered version of
// its module (design 47 G §374, A-68), as heardToolchainAgreed keeps its event.
func heardVersionDelivered(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) {
var d meshwants.VersionDeliveredEvent
if err := json.Unmarshal(m.Data(), &d); err != nil {
logf("%s holds what is no version-delivered, left: %v", m.Subject(), err)
_ = m.Ack()
return
}
if err := ix.HearVersionDelivered(ctx, d); err != nil {
logf("versions: the version-delivered of %s %s could not be kept, so the bus gives it again: %v", d.Module, d.Version, err)
_ = m.Nak()
return
}
_ = m.Ack()
}
// sayVersionsFilled says versions-filled (design 47 G §372) once, on the seat's stream that keeps it with no age limit.
func sayVersionsFilled(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.VersionsFilledEvent, string) error {
return func(_ context.Context, e meshcatalog.VersionsFilledEvent, id string) error {
@@ -113,6 +113,8 @@ type Indexer struct {
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.
SayVersionsFilled func(ctx context.Context, e meshcatalog.VersionsFilledEvent, id string) error
// Inputs is the state inputs (run_inputs.go, design 47 G §354): the inputs of the indexer's run asks; nil keeps none.
Inputs Store
// Registry is the registrations and their trusted changes (registrations.go, design 47 G §2 to §4, §667); SeedModules reads
// what the first registrations are composed from (§8). Nil: no registration act is answered.
Registry *Registry
@@ -910,7 +910,7 @@ func (ix *Indexer) dry() *dryRun {
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
OwnBuildState: nil, underSeats: true,
// A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep.
Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil,
Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil,
// A dry run writes no registration and asks no one.
Registry: nil, SeedModules: nil}
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
@@ -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", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled"} {
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -0,0 +1,181 @@
package index
import (
"context"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"slices"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// The indexer's inputs (design 47 G §354): the input of the ask `run` the indexer's validation puts, kept in the old state
// section inputs, one key per input named by its write time and its sha256, read by the seat mesh-catalog's verb input (whose
// callers are the holders of the build seats) and by the tool indexer_inputs. A key older than a day is deleted at each new
// ask, judged over at most the 20 oldest keys.
const (
InputsState = "inputs"
InputsBucket = Module + "_" + InputsState
InputsTool = "indexer_inputs"
// InputsKeptFor is how long an input is kept at the longest.
InputsKeptFor = 24 * time.Hour
// InputsSwept is how many of the oldest keys each new ask judges.
InputsSwept = 20
inputsTime = "20060102T150405Z"
)
// InputDigest is the sha256 of an input's exact bytes, as "sha256:" and its hex digits.
func InputDigest(body []byte) string {
sum := sha256.Sum256(body)
return "sha256:" + hex.EncodeToString(sum[:])
}
// KeepInput keeps the exact bytes of a run ask's input under their write time and sha256, after sweeping the keys more than a
// day old among the 20 oldest; it answers the digest. The same bytes written again in the same second are one key.
func (ix *Indexer) KeepInput(ctx context.Context, body []byte) (string, error) {
if ix.Inputs == nil {
return "", errors.New("the state inputs is not kept here")
}
if err := ix.sweepInputs(ctx); err != nil {
ix.logf("inputs: the keys older than a day are not all deleted: %v", err)
}
digest := InputDigest(body)
key := meshcatalog.InputsByTimeDigestKey(ix.now().UTC().Format(inputsTime), digest)
data, err := json.Marshal(base64.StdEncoding.EncodeToString(body))
if err != nil {
return "", err
}
if err := ix.Inputs.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) {
return "", err
}
return digest, nil
}
// sweepInputs deletes, among the 20 oldest keys, those written more than a day ago.
func (ix *Indexer) sweepInputs(ctx context.Context) error {
all, err := ix.Inputs.Keys(ctx)
if err != nil {
return err
}
slices.Sort(all) // the write time leads the key, so the oldest come first
if len(all) > InputsSwept {
all = all[:InputsSwept]
}
var errs []error
for _, k := range all {
t, _, ok := meshcatalog.ParseInputsByTimeDigestKey(k)
at, perr := time.Parse(inputsTime, t)
if !ok || perr != nil || ix.now().Sub(at) < InputsKeptFor {
continue
}
if _, rev, found, err := ix.Inputs.Get(ctx, k); err == nil && found {
if err := ix.Inputs.Delete(ctx, k, rev); err != nil {
errs = append(errs, err)
}
}
}
return errors.Join(errs...)
}
// InputKey is the key of the input of the digest, or none.
func InputKey(ctx context.Context, st Store, digest string) (string, error) {
all, err := st.Keys(ctx)
if err != nil {
return "", err
}
for _, k := range all {
if _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k); ok && d == digest {
return k, nil
}
}
return "", nil
}
// Input answers the verb input: the exact bytes (base64) of the input of the digest, its digest checked.
func Input(ctx context.Context, st Store, in meshcatalog.InputInput) (meshcatalog.InputAnswer, error) {
var out meshcatalog.InputAnswer
if in.Digest == "" {
return out, errors.New("name the input's sha256 (digest)")
}
key, err := InputKey(ctx, st, in.Digest)
if err != nil {
return out, fmt.Errorf("the state %s cannot be read: %w", InputsState, err)
}
raw, _, found, err := st.Get(ctx, key)
if key == "" || err != nil || !found {
return out, fmt.Errorf("no input of digest %s is kept", in.Digest)
}
text, err := meshcatalog.DecodeInputsByTimeDigestValue(raw)
if err != nil {
return out, fmt.Errorf("the input %s cannot be read: %w", key, err)
}
body, err := base64.StdEncoding.DecodeString(text)
if err != nil || InputDigest(body) != in.Digest {
return out, fmt.Errorf("the input kept under %s does not match its digest", key)
}
return meshcatalog.InputAnswer{Digest: in.Digest, Input: text}, nil
}
// InputTool is the verb input as the tool mesh-catalog.input of the indexer's own runtime.
func InputTool(st Store) stdio.Tool {
return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbInput, Run: func(args map[string]any) (any, error) {
var in meshcatalog.InputInput
if raw, err := json.Marshal(args); err == nil {
_ = json.Unmarshal(raw, &in)
}
return Input(context.Background(), st, in)
}}
}
// InputsAnswer is what indexer_inputs answers: each key kept with its time, digest and size, or one input's bytes.
type InputsAnswer struct {
Inputs []InputKept `json:"inputs"`
}
// InputKept is one input kept.
type InputKept struct {
Key string `json:"key"`
Digest string `json:"digest"`
Bytes int `json:"bytes"`
Input string `json:"input,omitempty"`
}
// InputsReadTool is indexer_inputs (class read): every input the indexer keeps, with its key, digest and size; with a digest,
// that input whole (base64).
func InputsReadTool(st Store) stdio.Tool {
return stdio.Tool{Name: InputsTool, Description: "Read the indexer's inputs of the ask run (design 47 G section 354): each key (write time and sha256) with its size; with a digest, that input's exact bytes, base64.",
Input: map[string]any{"type": "object", "properties": map[string]any{"digest": map[string]any{"type": "string", "description": "one input's sha256; none lists every key"}}},
Run: func(args map[string]any) (any, error) {
ctx := context.Background()
want, _ := args["digest"].(string)
all, err := st.Keys(ctx)
if err != nil {
return nil, err
}
slices.Sort(all)
out := InputsAnswer{Inputs: []InputKept{}}
for _, k := range all {
_, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k)
if !ok || (want != "" && d != want) {
continue
}
raw, _, found, err := st.Get(ctx, k)
if err != nil || !found {
continue
}
text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw)
kept := InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))}
if want != "" {
kept.Input = text
}
out.Inputs = append(out.Inputs, kept)
}
return out, nil
}}
}
@@ -1,6 +1,7 @@
package index
import (
"cmp"
"context"
"encoding/json"
"errors"
@@ -32,8 +33,9 @@ const (
// published is what a not yet released version holds: each artifact published at it, by reference.
type published struct {
Artifacts map[string]meshcatalog.VersionArtifact `json:"artifacts"`
Source int64 `json:"source,omitempty"`
Artifacts map[string]meshcatalog.VersionArtifact `json:"artifacts"`
Dependencies []meshbuilds.Dependency `json:"dependencies,omitempty"`
Source int64 `json:"source,omitempty"`
}
// versionParts is MAJOR.MINOR.PATCH, read; false for any other form.
@@ -113,10 +115,14 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag
if _, _, found, err := ix.Versions.Get(ctx, meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)); err != nil || found {
return err // released already: heard again
}
want, err := ix.artifactsOf(ctx, p.Module)
entry, err := ix.entryOf(ctx, p.Module)
if err != nil {
return err
}
var want []string
for _, a := range entry.Artifacts {
want = append(want, a.Ref)
}
key := PendingPrefix + meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)
var have published
err = write(ctx, ix.OwnBuildState, key, func(raw []byte, found bool) ([]byte, bool, error) {
@@ -131,6 +137,11 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag
}
have.Artifacts[p.Artifact] = meshcatalog.VersionArtifact{Artifact: p.Artifact, Digest: p.Digest}
have.Source = p.Source
for _, d := range p.Dependencies {
if !slices.ContainsFunc(have.Dependencies, func(x meshbuilds.Dependency) bool { return x.Kind == d.Kind && x.Part == d.Part }) {
have.Dependencies = append(have.Dependencies, d)
}
}
data, err := json.Marshal(have)
return data, false, err
})
@@ -142,41 +153,36 @@ func (ix *Indexer) HearPackagePublished(ctx context.Context, p meshbuilds.Packag
return nil
}
}
return ix.releaseVersion(ctx, p, key, want, have)
return ix.releaseVersion(ctx, p, key, entry, have)
}
// artifactsOf are the artifact references the module's entry in the newest trunk revision names.
func (ix *Indexer) artifactsOf(ctx context.Context, module string) ([]string, error) {
// entryOf is the module's entry in the newest trunk revision.
func (ix *Indexer) entryOf(ctx context.Context, module string) (meshcatalog.CatalogEntry, error) {
if ix.Revisions == nil {
return nil, errors.New("the catalog graph is not kept here: no entry names the module's artifacts")
return meshcatalog.CatalogEntry{}, errors.New("the catalog graph is not kept here: no entry names the module's artifacts")
}
rev, found, err := newestTrunk(ctx, ix.Revisions)
if err != nil {
return nil, err
return meshcatalog.CatalogEntry{}, err
}
if !found {
return nil, errors.New("the indexer keeps no trunk revision yet")
return meshcatalog.CatalogEntry{}, errors.New("the indexer keeps no trunk revision yet")
}
_, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision))
if err != nil {
return nil, err
return meshcatalog.CatalogEntry{}, err
}
for _, e := range entries {
if e.Module != module {
continue
if e.Module == module {
return e, nil
}
var refs []string
for _, a := range e.Artifacts {
refs = append(refs, a.Ref)
}
return refs, nil
}
return nil, fmt.Errorf("the trunk holds no module %s", module)
return meshcatalog.CatalogEntry{}, fmt.Errorf("the trunk holds no module %s", module)
}
// releaseVersion records the version of p whole, says version-released unless it is late, deletes what exceeds the kept versions,
// and forgets the artifacts kept as published.
func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePublished, pendingKey string, want []string, have published) error {
func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePublished, pendingKey string, entry meshcatalog.CatalogEntry, have published) error {
key := meshcatalog.VersionsByModuleVersionKey(p.Module, p.Version)
kept, err := ModuleVersions(ctx, ix.Versions, p.Module)
if err != nil {
@@ -185,9 +191,11 @@ func (ix *Indexer) releaseVersion(ctx context.Context, p meshbuilds.PackagePubli
late := len(kept) > 0 && compareVersions(kept[0].Version, p.Version) > 0
at := ix.now().UTC()
v := meshcatalog.VersionEntry{V: int64(keys.Version), Module: p.Module, Version: p.Version, Source: have.Source, At: &at, Late: late}
for _, ref := range slices.Sorted(slices.Values(want)) {
v.Artifacts = append(v.Artifacts, have.Artifacts[ref])
for _, a := range entry.Artifacts {
v.Artifacts = append(v.Artifacts, have.Artifacts[a.Ref])
}
slices.SortFunc(v.Artifacts, func(a, b meshcatalog.VersionArtifact) int { return strings.Compare(a.Artifact, b.Artifact) })
v.Edges = releaseEdges(entry, have.Dependencies)
data, err := json.Marshal(v)
if err != nil {
return err
@@ -270,3 +278,46 @@ func VersionsTool(st Store) stdio.Tool {
return Versions(context.Background(), st, in)
}}
}
// notYetDelivered is the version of an edge entry of kind 2, 4 or 6 until its release is first delivered (design 47 G §374).
const notYetDelivered = "not yet delivered"
// releaseEdges are the edge entries of a release (design 47 G §374): kinds 1 and 3 from the dependencies the build
// resolved, as package-published carried them (stale: the dependent is made from them, §378); kinds 2, 4 and 6 of the entry's
// catalog graph edges, reading "not yet delivered" until the first delivery writes them. Kind 5 is not written here.
func releaseEdges(entry meshcatalog.CatalogEntry, deps []meshbuilds.Dependency) []meshcatalog.VersionEdge {
word := func(kind int64, part string) string {
for _, e := range entry.Edges {
if e.Kind == kind && (e.Part == part || e.From == part) {
return e.Word
}
}
return ""
}
var out []meshcatalog.VersionEdge
for _, d := range deps {
out = append(out, meshcatalog.VersionEdge{Kind: d.Kind, Part: d.Part, Digest: d.Digest, Version: d.Version, Stale: d.Kind == 1 || d.Kind == 3, Word: word(d.Kind, d.Part)})
}
for _, e := range entry.Edges {
if e.Kind != 2 && e.Kind != 4 && e.Kind != 6 {
continue
}
part := cmp.Or(e.Part, e.From)
if !slices.ContainsFunc(out, func(x meshcatalog.VersionEdge) bool { return x.Kind == e.Kind && x.Part == part && x.Word == e.Word }) {
out = append(out, meshcatalog.VersionEdge{Kind: e.Kind, Part: part, Word: e.Word, Version: notYetDelivered})
}
}
return out
}
// OutdatedTool is the verb outdated as the tool mesh-catalog.outdated of the indexer's own runtime: it reads the versions
// state and the heard toolchain and delivered keys of the own state.
func OutdatedTool(ix *Indexer) stdio.Tool {
return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbOutdated, Run: func(args map[string]any) (any, error) {
var in meshcatalog.OutdatedInput
if raw, err := json.Marshal(args); err == nil {
_ = json.Unmarshal(raw, &in)
}
return ix.Outdated(context.Background(), in)
}}
}
@@ -0,0 +1,197 @@
package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"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-delivery/contracts/go/seats/meshwants"
)
// What outdated compares against is kept as heard, in the indexer's own state (design 47 G A-68): toolchain.<language> the newest
// toolchain-agreed per language, delivered.<module> the newest version-delivered per module.
const (
ToolchainPrefix = "toolchain."
DeliveredPrefix = "delivered."
)
// keepNewest writes body at key of the own-build state unless it holds a value whose version is higher; version of the body
// is read by newer, which says whether the heard one is not older than the kept.
func (ix *Indexer) keepNewest(ctx context.Context, key string, body any, newer func(kept []byte) bool) error {
if ix.OwnBuildState == nil {
return errors.New("the own state is not kept here")
}
data, err := json.Marshal(body)
if err != nil {
return err
}
return write(ctx, ix.OwnBuildState, key, func(kept []byte, found bool) ([]byte, bool, error) {
if found && !newer(kept) {
return nil, false, nil
}
return data, false, nil
})
}
// HearToolchainAgreed keeps the newest toolchain-agreed of a language (A-68, §375): the one heard last.
func (ix *Indexer) HearToolchainAgreed(ctx context.Context, t meshbuilds.ToolchainAgreed) error {
if t.Language == "" {
return errors.New("a toolchain-agreed names no language")
}
return ix.keepNewest(ctx, ToolchainPrefix+t.Language, t, func([]byte) bool { return true })
}
// HearVersionDelivered takes one version-delivered (design 47 G §374, §376, A-68): the entries of kinds 2, 4 and 6 of the
// delivered release, which read "not yet delivered", are written once with each dependency's newest delivered version and
// part digest; then the delivered version is kept as the module's newest.
func (ix *Indexer) HearVersionDelivered(ctx context.Context, d meshwants.VersionDeliveredEvent) error {
if ix.Versions == nil || ix.OwnBuildState == nil {
return errors.New("the state versions is not kept here")
}
if _, ok := versionParts(d.Version); !ok {
return fmt.Errorf("the version-delivered of %s names the version %q, which is no MAJOR.MINOR.PATCH", d.Module, d.Version)
}
key := meshcatalog.VersionsByModuleVersionKey(d.Module, d.Version)
err := write(ctx, ix.Versions, key, func(raw []byte, found bool) ([]byte, bool, error) {
if !found {
return nil, false, nil // a version the catalog does not hold (a build of another branch) has no entries to write
}
v, err := meshcatalog.DecodeVersionsByModuleVersionValue(raw)
if err != nil {
return nil, false, err
}
changed := false
for i, e := range v.Edges {
if e.Version != notYetDelivered {
continue
}
dep, ok := ix.deliveredOf(ctx, strings.SplitN(e.Part, ":", 2)[0])
if !ok {
continue // the dependency is not delivered yet: the entry reads so until this release's own delivery finds it
}
v.Edges[i].Version, v.Edges[i].Digest = dep.Version, dep.Parts[e.Part]
changed = true
}
if !changed {
return nil, false, nil
}
data, err := json.Marshal(v)
return data, false, err
})
if err != nil {
return err
}
return ix.keepNewest(ctx, DeliveredPrefix+d.Module, d, func(kept []byte) bool {
var k meshwants.VersionDeliveredEvent
return json.Unmarshal(kept, &k) != nil || compareVersions(d.Version, k.Version) >= 0
})
}
// deliveredOf is the newest version-delivered kept of a module.
func (ix *Indexer) deliveredOf(ctx context.Context, module string) (meshwants.VersionDeliveredEvent, bool) {
var d meshwants.VersionDeliveredEvent
raw, _, found, err := ix.OwnBuildState.Get(ctx, DeliveredPrefix+module)
if err != nil || !found || json.Unmarshal(raw, &d) != nil {
return d, false
}
return d, true
}
// Outdated answers the verb outdated (design 47 G §375) for the edge kinds 1, 2 and 4: every entry of each module's newest
// released version that is behind by the part's digest: kind 1, the recorded digest differs from the part's digest in the
// dependency's newest released version; kinds 2 and 4, from the dependency's newest version-delivered. A stale edge (kind 1)
// has the act rebuild, an ordering edge is shown and starts nothing. A dependency's newer version that left the digest as it
// was makes nothing behind.
func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (meshcatalog.OutdatedAnswer, error) {
out := meshcatalog.OutdatedAnswer{Behind: []meshcatalog.Outdated{}}
if ix.Versions == nil || ix.OwnBuildState == nil {
return out, errors.New("the state versions is not kept here")
}
all, err := ix.Versions.Keys(ctx)
if err != nil {
return out, err
}
modules := map[string]bool{}
for _, k := range all {
if m, _, ok := meshcatalog.ParseVersionsByModuleVersionKey(k); ok {
modules[m] = true
}
}
newest := map[string]meshcatalog.VersionEntry{}
for m := range modules {
vs, err := ModuleVersions(ctx, ix.Versions, m)
if err != nil {
return out, err
}
for _, v := range vs {
if !v.Late {
newest[m] = v
break
}
}
}
for _, m := range sortedKeys(newest) {
v := newest[m]
for _, e := range v.Edges {
dep := strings.SplitN(e.Part, ":", 2)[0]
if in.Dependency != "" && in.Dependency != e.Part {
continue
}
var latest, digest string
switch e.Kind {
case 1:
n, ok := newest[dep]
if !ok {
continue
}
for _, a := range n.Artifacts {
if a.Artifact == e.Part {
latest, digest = n.Version, a.Digest
}
}
case 2, 4:
d, ok := ix.deliveredOf(ctx, dep)
if !ok || e.Version == notYetDelivered {
continue
}
latest, digest = d.Version, d.Parts[e.Part]
default:
continue
}
if digest == "" || digest == e.Digest {
continue
}
o := meshcatalog.Outdated{Module: m, Version: v.Version, Kind: e.Kind, Recorded: e.Version + " " + e.Digest, Newest: latest + " " + digest,
Dependency: meshcatalog.OutdatedDependency{Artifact: e.Part, Branch: ix.firstBranch(ctx, dep)}}
if e.Stale {
o.Act = "rebuild"
}
out.Behind = append(out.Behind, o)
}
}
return out, nil
}
// firstBranch is the first watched branch of a module's registration, which its released versions are of.
func (ix *Indexer) firstBranch(ctx context.Context, module string) string {
if ix.Registry != nil {
if r, found, err := ix.Registry.Registration(ctx, module); err == nil && found && len(r.Branches) > 0 {
return r.Branches[0]
}
}
return ""
}
func sortedKeys[V any](m map[string]V) []string {
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
slices.Sort(out)
return out
}
@@ -4,11 +4,13 @@ import (
"context"
"errors"
"fmt"
"strings"
"testing"
"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-delivery/contracts/go/seats/meshwants"
)
// Design 47 G §349: a version is released once every artifact the module's entry names is published at it, said once as
@@ -157,3 +159,126 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
t.Errorf("filled again: %v, said %d", err, len(filledSaid))
}
}
// Design 47 G §374, §375, A-68: a release records, per edge, the part it was built against (kind 1, from package-published's
// dependencies) or, once delivered, the delivered version it was ordered after (kinds 2 and 4); outdated answers an edge behind
// by the part's digest alone, with the act rebuild only for a stale edge; a newer version that left the digest as it was
// makes nothing behind.
func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Revisions: newMemObjects(1 << 20)}
entries := map[string]meshcatalog.CatalogEntry{
"lib": {Module: "lib", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "lib:pkg"}}},
"dep": {Module: "dep", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "dep:program"}}},
"app": {Module: "app", 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"},
{Kind: 4, From: "dep:program", To: "app:program", Part: "dep:program", Word: "provision"}}},
}
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, entries)
if err != nil {
t.Fatal(err)
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
t.Fatal(err)
}
pub := func(module, version, digest string, deps ...meshbuilds.Dependency) {
t.Helper()
if err := ix.HearPackagePublished(ctx, meshbuilds.PackagePublished{Module: module, Artifact: module + ":" + map[string]string{"lib": "pkg", "dep": "program", "app": "program"}[module], Digest: digest, Version: version, Dependencies: deps}); err != nil {
t.Fatal(err)
}
}
behind := func() []meshcatalog.Outdated {
t.Helper()
a, err := ix.Outdated(ctx, meshcatalog.OutdatedInput{})
if err != nil {
t.Fatal(err)
}
return a.Behind
}
pub("lib", "0.1.1", "d1")
pub("dep", "0.1.1", "x1")
pub("app", "0.1.1", "a1", meshbuilds.Dependency{Kind: 1, Part: "lib:pkg", Version: "0.1.1", Digest: "d1"})
v, _ := ModuleVersions(ctx, ix.Versions, "app")
if len(v[0].Edges) != 2 || v[0].Edges[0].Kind != 1 || !v[0].Edges[0].Stale || v[0].Edges[0].Word != "imports" || v[0].Edges[1].Version != "not yet delivered" {
t.Fatalf("the release's edges: %+v", v[0].Edges)
}
pub("lib", "0.1.2", "d1") // a newer version, the part's digest as it was
if b := behind(); len(b) != 0 {
t.Fatalf("a newer version that left the digest as it was is behind: %+v", b)
}
pub("lib", "0.1.3", "d3")
if b := behind(); len(b) != 1 || b[0].Module != "app" || b[0].Kind != 1 || b[0].Act != "rebuild" || b[0].Dependency.Artifact != "lib:pkg" || !strings.Contains(b[0].Newest, "d3") || !strings.Contains(b[0].Recorded, "d1") {
t.Fatalf("behind after lib's new digest: %+v", b)
}
// delivery: dep 0.1.1 is delivered, then app 0.1.1 writes its kind 4 entry; dep 0.1.2 delivered later is behind, an ordering edge
deliver := func(module, version string, parts map[string]string) {
t.Helper()
if err := ix.HearVersionDelivered(ctx, meshwants.VersionDeliveredEvent{Module: module, Version: version, Parts: parts}); err != nil {
t.Fatal(err)
}
}
deliver("dep", "0.1.1", map[string]string{"dep:program": "x1"})
deliver("app", "0.1.1", map[string]string{"app:program": "a1"})
v, _ = ModuleVersions(ctx, ix.Versions, "app")
if e := v[0].Edges[1]; e.Version != "0.1.1" || e.Digest != "x1" {
t.Fatalf("the kind 4 entry after the first delivery: %+v", e)
}
deliver("dep", "0.1.2", map[string]string{"dep:program": "x2"})
deliver("app", "0.1.1", map[string]string{"app:program": "a1"}) // delivered again: written once, not changed
v, _ = ModuleVersions(ctx, ix.Versions, "app")
if e := v[0].Edges[1]; e.Version != "0.1.1" {
t.Errorf("an entry written at the first delivery changed: %+v", e)
}
b := behind()
if len(b) != 2 || b[1].Kind != 4 || b[1].Act != "" || b[1].Dependency.Artifact != "dep:program" {
t.Fatalf("behind after dep's second delivery: %+v", b)
}
// the newest toolchain-agreed per language is kept as heard
if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.25"}); err != nil {
t.Fatal(err)
}
if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.26"}); err != nil {
t.Fatal(err)
}
if raw, _, found, _ := ix.OwnBuildState.Get(ctx, ToolchainPrefix+"go"); !found || !strings.Contains(string(raw), "1.26") {
t.Errorf("toolchain.go holds %s", raw)
}
}
// Design 47 G §354: an input is kept under its write time and sha256, read back by the verb input with its digest checked;
// the keys older than a day among the 20 oldest are deleted at each new keep.
func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }}
d1, err := ix.KeepInput(ctx, []byte("first"))
if err != nil {
t.Fatal(err)
}
a, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1})
if err != nil || a.Digest != d1 || a.Input != "Zmlyc3Q=" {
t.Fatalf("the input read back: %+v, %v", a, err)
}
if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: "sha256:none"}); err == nil {
t.Error("an input never kept is answered")
}
now = now.Add(23 * time.Hour)
if _, err := ix.KeepInput(ctx, []byte("second")); err != nil {
t.Fatal(err)
}
if keys, _ := ix.Inputs.Keys(ctx); len(keys) != 2 {
t.Fatalf("inside a day: %v", keys)
}
now = now.Add(2 * time.Hour)
d3, _ := ix.KeepInput(ctx, []byte("third"))
keys, _ := ix.Inputs.Keys(ctx)
if len(keys) != 2 {
t.Fatalf("after a day the first stays: %v", keys)
}
if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil {
t.Error("the input a day old is still answered")
}
if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d3}); err != nil {
t.Error(err)
}
}
@@ -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}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17
{Name: index.VersionsState, History: 1, MaxMiB: 64}, {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,
@@ -159,9 +159,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
// internal/broker/state.go:62-93).
// The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*).
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments",
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments",
"mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.source-moved", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
"mesh-delivery.source-moved", "mesh-wants.version-delivered", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -196,8 +196,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered.
index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool,
index.EntryIndexTool, // design 47 G §675
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool) // design 47 G §667, §8
if !reflect.DeepEqual(m.Tools, want) || len(want) != 45 || slices.Contains(m.Tools, "modules") {
index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool) // design 47 G §667, §8
if !reflect.DeepEqual(m.Tools, want) || len(want) != 46 || slices.Contains(m.Tools, "modules") {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section.
@@ -301,7 +301,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// ADR 0604 §2, ADR 0636 §4: it holds the seat mesh-catalog, serving its four verbs and seat-definitions from its tools
// bundle (which stays without a request subject); ADR 0663 §1: and revision, in the same runtime of its own.
if len(m.Claims) != 1 || m.Claims[0].Name != index.CatalogSeat || m.Claims[0].Scope != catalogue.ScopeMesh ||
!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision", "entry", "registrations", "register", "policy", "watch", "unwatch", "versions"}) {
!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents", index.SeatDefinitionsVerb, "revision", "entry", "registrations", "register", "policy", "watch", "unwatch", "versions", "outdated", "input"}) {
t.Errorf("claims = %+v, want mesh-catalog with its seven verbs (entry: design 47 G §675)", m.Claims)
}
for _, e := range []string{index.SeatDefinitionChangedEvent, index.SeatDefinitionRemovedEvent} {
+9 -3
View File
@@ -20,7 +20,9 @@
"policy",
"watch",
"unwatch",
"versions"
"versions",
"outdated",
"input"
]
}
],
@@ -69,7 +71,8 @@
"indexer_entries",
"indexer_proposals",
"indexer_proposal_drop",
"indexer_seed_again"
"indexer_seed_again",
"indexer_inputs"
],
"consumes": [
"go-package-registry.published",
@@ -81,6 +84,8 @@
"mesh-delivery-checker.try.ended",
"mesh-delivery-checker.try.started",
"mesh-builds.package-published",
"mesh-builds.toolchain-agreed",
"mesh-wants.version-delivered",
"mesh-delivery-checker.verdict",
"mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.source-moved",
@@ -110,7 +115,8 @@
{ "name": "entry-index", "history": 1, "max-mib": 8 },
{ "name": "registrations", "history": 1 },
{ "name": "proposals", "history": 1, "max-mib": 4 },
{ "name": "versions", "history": 1, "max-mib": 64 }
{ "name": "versions", "history": 1, "max-mib": 64 },
{ "name": "inputs", "history": 1 }
],
"streams": [
{ "name": "registration-history", "events": ["registration.*"], "retention": "history", "max-mib": 64 },