The indexer keeps a catalog graph revision at each merge and each moved fingerprint, and says catalog-revision (milestone 47/G W4.3, its writing part)

Design 47 (Two graphs) keeps a catalog graph revision for each source revision taken in and for a fingerprint moved
with no merge. The recompute that writes the module graph now keeps the revision it makes, numbered after the newest
kept trunk revision, and says the seat mesh-catalog's event catalog-revision naming the changed modules; a recompute
that moves no entry keeps none, and objects not yet on the bus keep none, said in the log.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 11:45:07 +01:00
parent f0b51bfc29
commit bd067ae779
9 changed files with 177 additions and 22 deletions
@@ -435,6 +435,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: awaitedStore{b: graph},
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
Message: messageVia(calls.CallKept), Revisions: &busObjects{js: js, local: index.CatalogRevisionsObjects}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
SayCatalogRevision: sayCatalogRevision(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.
Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)},
@@ -3,13 +3,19 @@ package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-indexer/internal/index"
)
@@ -34,8 +40,11 @@ func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) {
return b.o, nil
}
o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(indexerModule, b.local))
if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
return nil, fmt.Errorf("the object bucket %s is not on the bus yet (the grants part creates it from the manifest's objects): %w", b.local, process.ErrNotYet)
}
if err != nil {
return nil, fmt.Errorf("the object bucket %s is not open (the grants part creates it from the manifest's objects): %w", b.local, err)
return nil, fmt.Errorf("the object bucket %s is not open: %w", b.local, err)
}
b.o = o
return o, nil
@@ -107,3 +116,24 @@ func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) {
}
return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil
}
// sayCatalogRevision publishes the seat mesh-catalog's event catalog-revision under the duplicate id of its kind and
// number, so a revision said twice is heard once (design 47, Two graphs).
func sayCatalogRevision(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.CatalogRevisionEvent) error {
return func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error {
data, err := json.Marshal(e)
if err != nil {
return err
}
msg := nats.NewMsg(seatEventSubject(meshcatalog.EventCatalogRevision))
id := fmt.Sprintf("%s-%s-%d%s", meshcatalog.EventCatalogRevision, e.Kind, e.Revision, e.Heads)
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-source", cred.Module)
msg.Header.Set("x-node", cred.Node)
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
msg.Header.Set(nats.MsgIdHdr, id)
msg.Data = data
return conn.PublishMsg(msg)
}
}
@@ -100,6 +100,8 @@ type Indexer struct {
// Revisions is the object bucket catalog-revisions (catalog_revisions.go; design 47, Two graphs); nil keeps no
// catalog graph revision.
Revisions Objects
// SayCatalogRevision says the seat mesh-catalog's event catalog-revision for a revision kept; nil says nothing.
SayCatalogRevision func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error
// Graph is the module graph's state (GraphBucket, graph.go; novox/hq ADR 0603 §2); nil writes no graph.
Graph Store
// Files asks the forge for every blob under each directory at a commit, by directory then path relative to it, each
@@ -5,10 +5,13 @@ import (
"encoding/json"
"errors"
"fmt"
"slices"
"strconv"
"strings"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"git.novox.be/novox/mesh-sdk/go/process"
)
// The indexer's object bucket catalog-revisions (design 47, Two graphs; novox/hq ADR 0619 §2: declared in the old
@@ -118,3 +121,96 @@ func ReadRevision(ctx context.Context, o Objects, name string) (meshcatalog.Cata
}
return rev, entries, nil
}
// newestTrunk is the newest trunk revision the bucket keeps, its list only; none is a zero revision and false.
func newestTrunk(ctx context.Context, o Objects) (meshcatalog.CatalogRevision, bool, error) {
names, err := o.Names(ctx)
if err != nil {
return meshcatalog.CatalogRevision{}, false, err
}
var newest int64
for _, n := range names {
if r, ok := TrunkRevisionOf(n); ok && r > newest {
newest = r
}
}
if newest == 0 {
return meshcatalog.CatalogRevision{}, false, nil
}
data, err := o.Get(ctx, "trunk."+strconv.FormatInt(newest, 10))
if err != nil {
return meshcatalog.CatalogRevision{}, false, err
}
var rev meshcatalog.CatalogRevision
if err := json.Unmarshal(data, &rev); err != nil {
return rev, false, fmt.Errorf("trunk.%d: %w", newest, err)
}
return rev, true, nil
}
// changedEntries are the modules whose entry differs between two revisions' lists, one gone or added included, by name.
func changedEntries(before, after []meshcatalog.EntryRef) []string {
was := map[string]string{}
for _, r := range before {
was[r.Module] = r.Digest
}
var changed []string
for _, r := range after {
if d, ok := was[r.Module]; !ok || d != r.Digest {
changed = append(changed, r.Module)
}
delete(was, r.Module)
}
for m := range was {
changed = append(changed, m)
}
slices.Sort(changed)
return changed
}
// keepTrunkRevision keeps the catalog graph revision a merge's or a recompute's module graph makes (design 47, Two
// graphs: a revision for each source revision taken in, and for a fingerprint moved with no merge), numbered after the
// newest kept, and says catalog-revision; a graph whose entries are the newest revision's makes none. A revision not
// kept is said in the log and returned, so the take-in that made it is not said taken.
func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graphEdges, whole graph.Graph) error {
if ix.Revisions == nil {
return nil
}
entries := make(map[string]meshcatalog.CatalogEntry, len(edges))
for name, e := range edges {
entries[name] = CatalogEntryOf(name, e.cur, whole[name])
}
ix.mu.Lock()
src := int64(ix.source.rev)
ix.mu.Unlock()
newest, any, err := newestTrunk(ctx, ix.Revisions)
if errors.Is(err, process.ErrNotYet) {
ix.logf("catalog graph: no revision is kept: its objects %s are not on the bus yet", CatalogRevisionsObjects)
return nil
}
if err != nil {
ix.logf("catalog graph: no revision is kept: the newest is not read: %v", err)
return fmt.Errorf("the newest catalog graph revision: %w", err)
}
at := ix.now().UTC()
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: newest.Revision + 1, Kind: "trunk", Source: src, At: &at}, entries)
if err != nil {
return err
}
changed := changedEntries(newest.Entries, rev.Entries)
if any && len(changed) == 0 {
return nil
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err)
return err
}
ix.logf("catalog graph: revision %d of source revision %d kept, %d entries, %d changed", rev.Revision, src, len(rev.Entries), len(changed))
if ix.SayCatalogRevision != nil {
if err := ix.SayCatalogRevision(ctx, meshcatalog.CatalogRevisionEvent{V: CatalogEntryLayout, Revision: rev.Revision, Kind: rev.Kind,
Source: src, At: &at, Changed: changed, Entries: int64(len(rev.Entries))}); err != nil {
ix.logf("catalog graph: revision %d is not said: %v", rev.Revision, err)
}
}
return nil
}
@@ -81,22 +81,7 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges)
GraphMost, len(names), names[GraphMost], len(names)-GraphMost-1)
names = names[:GraphMost]
}
providers := func(provision string) []catalogue.Manifest {
var out []catalogue.Manifest
for _, name := range slices.Sorted(maps.Keys(edges)) {
if slices.Contains(edges[name].manifest.Offers(), provision) {
out = append(out, edges[name].manifest)
}
}
return out
}
// The verb edges on a cycle are left out over the whole graph (ADR 0614 §3), every module it knows, so a module past
// GraphMost still closes the cycle it lies on.
whole := graph.Graph{}
for name, e := range edges {
whole[name] = graphNode(e, providers)
}
whole.LeaveOut()
whole := wholeGraph(edges)
for _, name := range names {
key, err := graph.Key(name)
if err != nil {
@@ -141,3 +126,23 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges)
}
return nil
}
// wholeGraph is the module graph of every module of edges, the verb edges on a cycle left out over the whole graph
// (ADR 0614 §3), so a module past GraphMost still closes the cycle it lies on.
func wholeGraph(edges map[string]graphEdges) graph.Graph {
providers := func(provision string) []catalogue.Manifest {
var out []catalogue.Manifest
for _, name := range slices.Sorted(maps.Keys(edges)) {
if slices.Contains(edges[name].manifest.Offers(), provision) {
out = append(out, edges[name].manifest)
}
}
return out
}
whole := graph.Graph{}
for name, e := range edges {
whole[name] = graphNode(e, providers)
}
whole.LeaveOut()
return whole
}
@@ -886,7 +886,7 @@ func (ix *Indexer) dry() *dryRun {
Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: 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}
Revisions: nil, SayCatalogRevision: nil}
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
x.Store, d.candidates = over(ix.Store)
x.Modules, d.modules = over(ix.Modules)
@@ -970,7 +970,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", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions"} {
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -372,7 +372,10 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc
if err := ix.writeOverlay(ctx, overlay, finals, w); err != nil {
return err
}
return ix.writeGraph(ctx, edges)
if err := ix.writeGraph(ctx, edges); err != nil {
return err
}
return ix.keepTrunkRevision(ctx, edges, wholeGraph(edges))
}
// writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has
@@ -332,16 +332,34 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { //
if _, _, err := ReadRevision(context.Background(), kept, "trunk.2"); err == nil || !strings.Contains(err.Error(), next.Entries[0].Module) {
t.Errorf("trunk.2 with a lost entry read without error naming it: %v", err)
}
// W4.3: a recompute keeps a trunk revision of the module graph and says catalog-revision; one that moves no entry keeps
// none; a module leaving makes the next revision, naming it changed.
ix.Revisions = newMemObjects(0)
var revisions []meshcatalog.CatalogRevisionEvent
ix.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error {
revisions = append(revisions, e)
return nil
}
if err := ix.Recompute(context.Background()); err != nil || len(revisions) != 1 || revisions[0].Revision != 1 || revisions[0].Kind != "trunk" ||
revisions[0].Entries != int64(len(mods.vals)) || len(revisions[0].Changed) != len(mods.vals) {
t.Fatalf("the first recompute said %+v (%v), want trunk revision 1 with every module", revisions, err)
}
graph := ix.Graph.(*memStore)
written := graph.writes
if err := ix.Recompute(context.Background()); err != nil || graph.writes != written {
t.Errorf("a recompute that moves nothing wrote the graph %d times (%v)", graph.writes-written, err)
if err := ix.Recompute(context.Background()); err != nil || graph.writes != written || len(revisions) != 1 {
t.Errorf("a recompute that moves nothing wrote the graph %d times and said %d revisions (%v)", graph.writes-written, len(revisions), err)
}
// A module the modules state no longer holds (a merge dropped it, issue 1900) leaves the graph at the next recompute.
delete(mods.vals, "other")
if err := ix.Recompute(context.Background()); err != nil {
t.Fatal(err)
}
if len(revisions) != 2 || revisions[1].Revision != 2 || !reflect.DeepEqual(revisions[1].Changed, []string{"other"}) {
t.Errorf("after other left, said %+v, want revision 2 naming other changed", revisions)
}
if _, kept, err := ReadRevision(context.Background(), ix.Revisions, "trunk.2"); err != nil || len(kept) != len(mods.vals) {
t.Errorf("trunk.2 holds %d entries (%v), want %d", len(kept), err, len(mods.vals))
}
if _, held := graph.vals["other"]; held {
t.Error("graph.other is still held after its module went")
}