indexer: outdated for kinds 3, 5 and 6; the acts of wait-ended kept as heard

Builds: design 47 G A-78 a, b, A-80, §374, §375
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:13:38 +01:00
parent bec5e063bc
commit c6af94fe58
8 changed files with 319 additions and 17 deletions
@@ -289,7 +289,7 @@ func main() {
// 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)}),
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}}),
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}, Revisions: revisions}),
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"})
@@ -690,6 +690,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := hub.Handle(meshwants.SubjectVersionDelivered, func(m jetstream.Msg) { heardVersionDelivered(ctx, ix, m, logger.Printf) }); err != nil {
return err
}
if err := hub.Handle(meshwants.SubjectWaitEnded, func(m jetstream.Msg) { heardWaitEnded(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
}
@@ -94,6 +94,22 @@ func heardVersionDelivered(ctx context.Context, ix *index.Indexer, m jetstream.M
_ = m.Ack()
}
// heardWaitEnded keeps the act of an acting part's ended wait for kind 5 (A-78 a), as heardToolchainAgreed keeps its event.
func heardWaitEnded(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) {
var w meshwants.WaitEndedEvent
if err := json.Unmarshal(m.Data(), &w); err != nil {
logf("%s holds what is no wait-ended, left: %v", m.Subject(), err)
_ = m.Ack()
return
}
if err := ix.HearWaitEnded(ctx, w); err != nil {
logf("versions: the wait-ended of %s for %s could not be kept, so the bus gives it again: %v", w.Part, w.Module, 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 {
@@ -193,8 +193,8 @@ type Indexer struct {
// Heard keeps what the indexer heard and compares later (A-80): the time a call is unanswered since (steps-since, call-since),
// toolchain.<language>, delivered.<module>, pending.<module>.<version>, acted.<acting part>.<dependent>, each key part escaped as a name
// (§73), and the mark versions-filled; nil keeps none. It is the state heard, whose key forge keeps the forge messages handled.
Heard Store
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
Heard Store
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
// Message asks the forge bridge for the merged message of a merge, repository, merge commit and pull request number in (the number lets the bridge read a merge made on the forge, not through the gate), the message whole out (a
// pending value keeps a reference, not a copy: issue 1585); an error is the bridge's words. Nil asks none.
Message func(ctx context.Context, repository, merge string, number int) ([]byte, error)
@@ -1,12 +1,14 @@
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"
@@ -19,8 +21,59 @@ import (
const (
ToolchainPrefix = "toolchain."
DeliveredPrefix = "delivered."
// ActedPrefix opens acted.<acting part>.<dependent>, the newest wait-ended of an acting part's wait for a dependent, ran or released
// (A-78 a), each key part escaped as a name (A-80, §73).
ActedPrefix = "acted."
)
// Acted is what an acted key holds: the build the dependent was at when the acting part acted for it (the wait's version, §374 kind 5).
type Acted struct {
V int64 `json:"v"`
Part string `json:"part"`
Step string `json:"step"`
Build string `json:"build"`
How string `json:"how"`
At time.Time `json:"at"`
}
// HearWaitEnded keeps, for kind 5 (§374, §375, A-78 a), the newest wait-ended of an acting step that ended ran or released: the
// acting part, the step and the build it acted for. A wait that ended superseded, or of a step that is no acting part's (put,
// delivery, kept-build, backup), is not an act and keeps nothing.
func (ix *Indexer) HearWaitEnded(ctx context.Context, w meshwants.WaitEndedEvent) error {
if w.Part == "" || w.Module == "" || (w.How != "ran" && w.How != "released") {
return nil
}
switch w.Step {
case meshwants.WaitStepPut, meshwants.WaitStepDelivery, meshwants.WaitStepKeptBuild, meshwants.WaitStepBackup, "":
return nil
}
if ix.Heard == nil {
return errors.New("the state heard is not kept here")
}
a := Acted{V: 1, Part: w.Part, Step: w.Step, Build: w.Version, How: w.How, At: w.Since}
return ix.keepNewest(ctx, ActedKey(w.Part, w.Module), a, func(kept []byte) bool {
var k Acted
return json.Unmarshal(kept, &k) != nil || !a.At.Before(k.At)
})
}
// ActedKey is the key of the state heard of an acting part's act for a dependent.
func ActedKey(part, dependent string) string {
return ActedPrefix + keys.EscapeName(part) + "." + keys.EscapeName(dependent)
}
// actedFor is the act kept for the acting part (an artifact reference, or its module) and the dependent module.
func (ix *Indexer) actedFor(ctx context.Context, part, dependent string) (Acted, bool) {
for _, p := range []string{part, strings.SplitN(part, ":", 2)[0]} {
raw, _, found, err := ix.Heard.Get(ctx, ActedKey(p, dependent))
var a Acted
if err == nil && found && json.Unmarshal(raw, &a) == nil {
return a, true
}
}
return Acted{}, false
}
// keepNewest writes body at key of the state heard 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 {
@@ -103,11 +156,14 @@ func (ix *Indexer) deliveredOf(ctx context.Context, module string) (meshwants.Ve
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.
// Outdated answers the verb outdated (design 47 G §375) for every edge kind: every entry of each module's newest released
// version that is behind. Kind 1: the recorded digest differs from the part's digest in the dependency's newest released
// version. Kind 3: the recorded toolchain version differs from the version of the newest toolchain-agreed of the dependent
// artifact's language (A-80, toolchain.<language>). Kinds 2 and 4: from the dependency's newest version-delivered. Kind 5: the
// acting part has not acted for the dependent's current build (acted.<acting part>.<dependent> against the newest trunk
// entry's fingerprint, A-78 a). Kind 6: the delivered version's catalog entry no longer serves the verb or word (A-78 b). A
// stale edge (kinds 1, 3) 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.Heard == nil {
@@ -136,6 +192,16 @@ func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (
}
}
}
trunk := map[string]meshcatalog.CatalogEntry{} // the newest trunk revision's entries, read where a kind needs them
if ix.Revisions != nil {
if rev, found, err := newestTrunk(ctx, ix.Revisions); err == nil && found {
if _, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision)); err == nil {
for _, e := range entries {
trunk[e.Module] = e
}
}
}
}
for _, m := range sortedKeys(newest) {
v := newest[m]
for _, e := range v.Edges {
@@ -143,32 +209,62 @@ func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (
if in.Dependency != "" && in.Dependency != e.Part {
continue
}
var latest, digest string
o := meshcatalog.Outdated{Module: m, Version: v.Version, Kind: e.Kind,
Dependency: meshcatalog.OutdatedDependency{Artifact: e.Part, Branch: ix.firstBranch(ctx, dep)}}
switch e.Kind {
case 1:
n, ok := newest[dep]
if !ok {
continue
}
var digest string
for _, a := range n.Artifacts {
if a.Artifact == e.Part {
latest, digest = n.Version, a.Digest
digest = a.Digest
}
}
if digest == "" || digest == e.Digest {
continue
}
o.Recorded, o.Newest = e.Version+" "+e.Digest, n.Version+" "+digest
case 3:
tc, ok := ix.agreedToolchain(ctx, languageOf(trunk[m], e))
if !ok || tc.Version == "" || tc.Version == e.Version {
continue
}
o.Recorded, o.Newest = e.Version+" "+e.Digest, tc.Version
case 2, 4:
d, ok := ix.deliveredOf(ctx, dep)
if !ok || e.Version == notYetDelivered {
continue
}
latest, digest = d.Version, d.Parts[e.Part]
digest := d.Parts[e.Part]
if digest == "" || digest == e.Digest {
continue
}
o.Recorded, o.Newest = e.Version+" "+e.Digest, d.Version+" "+digest
case 5:
cur := trunk[m].Fp
if cur == "" {
continue
}
acted, ok := ix.actedFor(ctx, e.Part, m)
if ok && acted.Build == cur {
continue
}
o.Recorded, o.Newest = "acted for "+cmp.Or(acted.Build, "no build"), "the current build "+cur
case 6:
d, ok := ix.deliveredOf(ctx, dep)
if !ok || e.Version == notYetDelivered || d.Version == e.Version {
continue
}
if serves, known := ix.deliveredServes(ctx, dep, d.Version, e.Word); !known || serves {
continue
}
o.Recorded, o.Newest = e.Version+" "+e.Word, d.Version+" no longer serves "+e.Word
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"
}
@@ -178,6 +274,88 @@ func (ix *Indexer) Outdated(ctx context.Context, in meshcatalog.OutdatedInput) (
return out, nil
}
// agreedToolchain is the newest toolchain-agreed kept for a language (A-80).
func (ix *Indexer) agreedToolchain(ctx context.Context, language string) (meshbuilds.ToolchainAgreed, bool) {
var t meshbuilds.ToolchainAgreed
if language == "" {
return t, false
}
raw, _, found, err := ix.Heard.Get(ctx, ToolchainPrefix+keys.EscapeName(language))
if err != nil || !found || json.Unmarshal(raw, &t) != nil {
return t, false
}
return t, true
}
// languageOf is the language of the artifact a kind-3 edge of the entry builds: the target of the catalog graph's edge from the
// toolchain part, else the one language the entry's artifacts share.
func languageOf(entry meshcatalog.CatalogEntry, e meshcatalog.VersionEdge) string {
lang := map[string]string{}
for _, a := range entry.Artifacts {
lang[a.Ref] = a.Language
}
for _, g := range entry.Edges {
if g.Kind == 3 && (g.Part == e.Part || g.From == e.Part) && lang[g.To] != "" {
return lang[g.To]
}
}
one := ""
for _, l := range lang {
if l == "" || (one != "" && one != l) {
return ""
}
one = l
}
return one
}
// deliveredServes says whether the catalog entry of a delivered version of module serves word, its verb (seat.verb, as the entry's
// serves lists it, the form `seat:` allowed) or a word of its manifest (A-78 b); known is false where that version's entry cannot be read.
func (ix *Indexer) deliveredServes(ctx context.Context, module, version, word string) (serves, known bool) {
if ix.Revisions == nil {
return false, false
}
vs, err := ModuleVersions(ctx, ix.Versions, module)
if err != nil {
return false, false
}
var source int64
for _, v := range vs {
if v.Version == version {
source = v.Source
}
}
if source == 0 {
return false, false
}
names, err := ix.Revisions.Names(ctx)
if err != nil {
return false, false
}
var numbers []int64
for _, n := range names {
if r, ok := TrunkRevisionOf(n); ok {
numbers = append(numbers, r)
}
}
slices.SortFunc(numbers, func(a, b int64) int { return cmp.Compare(b, a) })
for _, n := range numbers { // the newest revision taken from the source revision the version was built at
rev, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", n))
if err != nil || rev.Source != source {
continue
}
for _, e := range entries {
if e.Module != module {
continue
}
w := strings.TrimPrefix(word, "seat:")
_, inManifest := e.Manifest[word]
return slices.Contains(e.Serves, w) || slices.Contains(e.Provides, w) || slices.Contains(e.Streams, w) || inManifest, true
}
}
return false, false
}
// 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 {
@@ -2,6 +2,7 @@ package index
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
@@ -307,3 +308,106 @@ func TestWhatIsHeardIsKeptEscapedInTheStateHeard(t *testing.T) {
t.Errorf("the own-build state holds %v", own)
}
}
// Design 47 G §375, A-78, A-80: outdated says a kind-3 edge behind when the agreed toolchain of its artifact's language names another
// version, a kind-5 edge when the acting part has not acted for the dependent's current build (a wait-ended that ended ran or
// released), and a kind-6 edge when the delivered version's entry no longer serves the word.
func TestOutdatedAnswersKinds3And5And6(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
keep := func(n, source int64, serves []string, fp string) {
t.Helper()
entries := map[string]meshcatalog.CatalogEntry{
"prov": {Module: "prov", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "prov:program"}}, Serves: serves},
"actor": {Module: "actor", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "actor:program"}}},
"app": {Module: "app", Manifest: map[string]any{}, Fp: fp, Artifacts: []meshcatalog.Artifact{{Ref: "app:program", Language: "go"}},
Edges: []meshcatalog.Edge{{Kind: 3, From: "tc:program", To: "app:program", Part: "tc:program"},
{Kind: 5, From: "actor:program", To: "app:program", Part: "actor:program"},
{Kind: 6, From: "prov:program", To: "app:program", Part: "prov:program", Word: "seat:s.verb"}}},
}
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: n, Source: source, Kind: "trunk"}, entries)
if err != nil {
t.Fatal(err)
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
t.Fatal(err)
}
}
keep(1, 10, []string{"s.verb"}, "fp1")
put := func(module string, v meshcatalog.VersionEntry) {
t.Helper()
v.Module, v.V = module, 1
data, _ := json.Marshal(v)
if err := ix.Versions.Create(ctx, meshcatalog.VersionsByModuleVersionKey(module, v.Version), data); err != nil {
t.Fatal(err)
}
}
put("prov", meshcatalog.VersionEntry{Version: "0.1.1", Source: 10})
put("app", meshcatalog.VersionEntry{Version: "0.1.1", Edges: []meshcatalog.VersionEdge{
{Kind: 3, Part: "tc:program", Version: "1.25", Digest: "t1", Stale: true},
{Kind: 5, Part: "actor:program"},
{Kind: 6, Part: "prov:program", Word: "seat:s.verb", Version: "0.1.1"}}})
behind := func() map[int64]meshcatalog.Outdated {
t.Helper()
a, err := ix.Outdated(ctx, meshcatalog.OutdatedInput{})
if err != nil {
t.Fatal(err)
}
m := map[int64]meshcatalog.Outdated{}
for _, o := range a.Behind {
m[o.Kind] = o
}
return m
}
if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.25"}); err != nil {
t.Fatal(err)
}
b := behind()
if _, ok := b[3]; ok {
t.Errorf("kind 3 behind at the agreed version: %+v", b[3])
}
if o, ok := b[5]; !ok || o.Act != "" {
t.Errorf("kind 5 is not behind before the acting part acted: %+v", b)
}
wait := func(how, step, build string, since time.Time) {
t.Helper()
if err := ix.HearWaitEnded(ctx, meshwants.WaitEndedEvent{Part: "actor:program", Module: "app", Step: step, How: how, Version: build, Since: since}); err != nil {
t.Fatal(err)
}
}
at := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
wait("superseded", "compose", "fp1", at)
wait("ran", "delivery", "fp1", at)
if _, ok := behind()[5]; !ok {
t.Error("a superseded wait or a delivery wait counted as an act")
}
wait("ran", "compose", "fp1", at)
if _, ok := behind()[5]; ok {
t.Error("kind 5 is behind after the act for the current build")
}
keep(2, 20, []string{"s.other"}, "fp2")
if o, ok := behind()[5]; !ok || !strings.Contains(o.Newest, "fp2") {
t.Errorf("kind 5 after the dependent's new build: %+v", o)
}
wait("released", "compose", "fp2", at.Add(time.Minute))
if _, ok := behind()[5]; ok {
t.Error("a released wait is not an act")
}
if err := ix.HearToolchainAgreed(ctx, meshbuilds.ToolchainAgreed{Language: "go", Version: "1.26"}); err != nil {
t.Fatal(err)
}
if o, ok := behind()[3]; !ok || o.Act != "rebuild" || o.Newest != "1.26" || !strings.Contains(o.Recorded, "1.25") {
t.Errorf("kind 3 after the agreed toolchain changed: %+v", o)
}
// kind 6: prov 0.1.2 built at source 20 serves another word only
if _, ok := behind()[6]; ok {
t.Error("kind 6 behind with no newer delivered version")
}
put("prov", meshcatalog.VersionEntry{Version: "0.1.2", Source: 20})
if err := ix.HearVersionDelivered(ctx, meshwants.VersionDeliveredEvent{Module: "prov", Version: "0.1.2", Parts: map[string]string{"prov:program": "p2"}}); err != nil {
t.Fatal(err)
}
if o, ok := behind()[6]; !ok || o.Act != "" || !strings.Contains(o.Newest, "no longer serves") {
t.Errorf("kind 6 after a version that serves no more: %+v", o)
}
}
@@ -161,7 +161,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// 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-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", "mesh-wants.version-delivered", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
"mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "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" {
@@ -86,6 +86,7 @@
"mesh-builds.package-published",
"mesh-builds.toolchain-agreed",
"mesh-wants.version-delivered",
"mesh-wants.wait-ended",
"mesh-delivery-checker.verdict",
"mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.source-moved",