The indexer keeps its catalog graph revisions in its object bucket catalog-revisions, each entry once by digest (milestone 47/G W4.2)

Design 47 (Two graphs) keeps each revision as the list of its entries' digests and each entry as one object, so an unchanged entry is stored once. The bucket is declared in the old objects section with a 512 MiB cap (ADR 0619 §2: the indexer is the foundation), opened at its first use; a revision's list is written only after its entries, and a revision read back with an entry lost is an error naming the module. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 11:45:07 +01:00
parent ab87109ec5
commit f0b51bfc29
10 changed files with 344 additions and 5 deletions
@@ -434,7 +434,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
 		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), // a kept call: a read, called again through a transport fault (ADR 0668 §1)
+		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)
 		// 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)},
@@ -0,0 +1,109 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+
+	"mesh-delivery-indexer/internal/index"
+)
+
+// indexerModule is the module this program runs as, whose object buckets it opens by their local names.
+const indexerModule = "mesh-delivery-indexer"
+
+// busObjects is one of the indexer's own object buckets in the old `objects` section (novox/hq ADR 0619 §2), which the
+// grants part creates: it is opened at its first use and kept; until the bucket exists every call is an error saying so,
+// and the next call opens it again (no timer: the next revision or tool call asks).
+type busObjects struct {
+	js    jetstream.JetStream
+	local string
+	mu    sync.Mutex
+	o     jetstream.ObjectStore
+}
+
+func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.o != nil {
+		return b.o, nil
+	}
+	o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(indexerModule, b.local))
+	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)
+	}
+	b.o = o
+	return o, nil
+}
+
+func (b *busObjects) Put(ctx context.Context, name string, data []byte) error {
+	o, err := b.open(ctx)
+	if err != nil {
+		return err
+	}
+	_, err = o.Put(ctx, jetstream.ObjectMeta{Name: name}, bytes.NewReader(data))
+	return err
+}
+
+func (b *busObjects) Get(ctx context.Context, name string) ([]byte, error) {
+	o, err := b.open(ctx)
+	if err != nil {
+		return nil, err
+	}
+	data, err := o.GetBytes(ctx, name)
+	if errors.Is(err, jetstream.ErrObjectNotFound) {
+		return nil, index.ErrNoObject
+	}
+	return data, err
+}
+
+func (b *busObjects) Delete(ctx context.Context, name string) error {
+	o, err := b.open(ctx)
+	if err != nil {
+		return err
+	}
+	if err := o.Delete(ctx, name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) {
+		return err
+	}
+	return nil
+}
+
+func (b *busObjects) Names(ctx context.Context) ([]string, error) {
+	o, err := b.open(ctx)
+	if err != nil {
+		return nil, err
+	}
+	infos, err := o.List(ctx)
+	if errors.Is(err, jetstream.ErrNoObjectsFound) {
+		return []string{}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	names := make([]string, 0, len(infos))
+	for _, i := range infos {
+		names = append(names, i.Name)
+	}
+	return names, nil
+}
+
+func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) {
+	o, err := b.open(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	st, err := o.Status(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	info, err := b.js.Stream(ctx, "OBJ_"+st.Bucket())
+	if err != nil {
+		return int64(st.Size()), 0, err
+	}
+	return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil
+}
@@ -97,6 +97,9 @@ type Indexer struct {
 	ReadFile func(ctx context.Context, repository, sha, path string) ([]byte, error)
 	// SaySeatDefinition says seat-definition-changed (seatdefs.go); nil says nothing.
 	SaySeatDefinition func(ctx context.Context, c SeatDefinitionChanged) error
+	// Revisions is the object bucket catalog-revisions (catalog_revisions.go; design 47, Two graphs); nil keeps no
+	// catalog graph revision.
+	Revisions Objects
 	// 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
@@ -0,0 +1,120 @@
+package index
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+)
+
+// The indexer's object bucket catalog-revisions (design 47, Two graphs; novox/hq ADR 0619 §2: declared in the old
+// `objects` section, its cap there): a revision is its list of entries' digests under trunk.<revision> or
+// candidate.<group>.<heads>, and each entry one object entry.<digest>, written once and shared by every revision that
+// lists it.
+
+// CatalogRevisionsObjects is the bucket's local name in the indexer's manifest.
+const CatalogRevisionsObjects = "catalog-revisions"
+
+// ErrNoObject is an object the bucket does not hold.
+var ErrNoObject = errors.New("no such object")
+
+// Objects is an object bucket as the indexer keeps its revisions in it.
+type Objects interface {
+	// Put stores data whole under name, replacing one of that name.
+	Put(ctx context.Context, name string, data []byte) error
+	// Get is the object's bytes; ErrNoObject when it holds none of that name.
+	Get(ctx context.Context, name string) ([]byte, error)
+	// Delete removes the object; one it does not hold is no error.
+	Delete(ctx context.Context, name string) error
+	// Names is every object's name; none is an empty list.
+	Names(ctx context.Context) ([]string, error)
+	// Fill is the bucket's bytes used and its cap in bytes.
+	Fill(ctx context.Context) (used, capacity int64, err error)
+}
+
+// EntryObject is the object name of the entry of that digest.
+func EntryObject(digest string) string { return "entry." + digest }
+
+// RevisionObject is the object name of a revision: trunk.<revision>, or candidate.<group>.<heads> for a candidate,
+// which keeps its newest revision only.
+func RevisionObject(rev meshcatalog.CatalogRevision) string {
+	if rev.Kind == "candidate" {
+		return "candidate." + rev.Group + "." + rev.Heads
+	}
+	return "trunk." + strconv.FormatInt(rev.Revision, 10)
+}
+
+// TrunkRevisionOf is the revision number a trunk object's name holds, and whether it is one.
+func TrunkRevisionOf(name string) (int64, bool) {
+	n, ok := strings.CutPrefix(name, "trunk.")
+	if !ok {
+		return 0, false
+	}
+	r, err := strconv.ParseInt(n, 10, 64)
+	return r, err == nil
+}
+
+// KeepRevision stores a revision as CatalogRevisionOf made it: each entry object the bucket does not hold yet, then the
+// revision's list, so a list is never kept whose entries are not; an entry already held is not written again.
+func KeepRevision(ctx context.Context, o Objects, rev meshcatalog.CatalogRevision, entries map[string][]byte) error {
+	held, err := o.Names(ctx)
+	if err != nil {
+		return fmt.Errorf("the objects %s: %w", CatalogRevisionsObjects, err)
+	}
+	have := make(map[string]bool, len(held))
+	for _, n := range held {
+		have[n] = true
+	}
+	for _, ref := range rev.Entries {
+		name := EntryObject(ref.Digest)
+		if have[name] {
+			continue
+		}
+		data, ok := entries[ref.Digest]
+		if !ok {
+			return fmt.Errorf("revision %s lists the entry of %s (%s), which it does not carry", RevisionObject(rev), ref.Module, ref.Digest)
+		}
+		if err := o.Put(ctx, name, data); err != nil {
+			return fmt.Errorf("the entry of %s in %s: %w", ref.Module, CatalogRevisionsObjects, err)
+		}
+		have[name] = true
+	}
+	list, err := json.Marshal(rev)
+	if err != nil {
+		return fmt.Errorf("revision %s: %w", RevisionObject(rev), err)
+	}
+	if err := o.Put(ctx, RevisionObject(rev), list); err != nil {
+		return fmt.Errorf("revision %s in %s: %w", RevisionObject(rev), CatalogRevisionsObjects, err)
+	}
+	return nil
+}
+
+// ReadRevision is the revision kept under name and its entries whole, in the revision's order; an entry it lists that
+// the bucket lost is an error naming it, never an entry left out.
+func ReadRevision(ctx context.Context, o Objects, name string) (meshcatalog.CatalogRevision, []meshcatalog.CatalogEntry, error) {
+	var rev meshcatalog.CatalogRevision
+	data, err := o.Get(ctx, name)
+	if err != nil {
+		return rev, nil, fmt.Errorf("revision %s: %w", name, err)
+	}
+	if err := json.Unmarshal(data, &rev); err != nil {
+		return rev, nil, fmt.Errorf("revision %s: %w", name, err)
+	}
+	entries := make([]meshcatalog.CatalogEntry, 0, len(rev.Entries))
+	for _, ref := range rev.Entries {
+		data, err := o.Get(ctx, EntryObject(ref.Digest))
+		if err != nil {
+			return rev, nil, fmt.Errorf("revision %s, the entry of %s (%s): %w", name, ref.Module, ref.Digest, err)
+		}
+		var e meshcatalog.CatalogEntry
+		if err := json.Unmarshal(data, &e); err != nil {
+			return rev, nil, fmt.Errorf("revision %s, the entry of %s: %w", name, ref.Module, err)
+		}
+		entries = append(entries, e)
+	}
+	return rev, entries, nil
+}
@@ -1,6 +1,12 @@
 package index

-import "encoding/json"
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"slices"
+	"sync"
+)

 // rawList is the list a kept predefined value holds, as it was written.
 func rawList(value []byte) json.RawMessage {
@@ -10,3 +16,69 @@ func rawList(value []byte) json.RawMessage {
 	_ = json.Unmarshal(value, &kept)
 	return kept.List
 }
+
+// memObjects is an object bucket in memory with a cap: a put past the cap is refused as the bus's DiscardNew refuses it.
+type memObjects struct {
+	mu       sync.Mutex
+	objs     map[string][]byte
+	puts     int
+	capacity int64
+}
+
+func newMemObjects(capacity int64) *memObjects {
+	return &memObjects{objs: map[string][]byte{}, capacity: capacity}
+}
+
+func (m *memObjects) used() int64 {
+	var n int64
+	for _, d := range m.objs {
+		n += int64(len(d))
+	}
+	return n
+}
+
+func (m *memObjects) Put(_ context.Context, name string, data []byte) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.capacity > 0 && m.used()-int64(len(m.objs[name]))+int64(len(data)) > m.capacity {
+		return errObjectsFullForTest
+	}
+	m.objs[name] = slices.Clone(data)
+	m.puts++
+	return nil
+}
+
+func (m *memObjects) Get(_ context.Context, name string) ([]byte, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	d, ok := m.objs[name]
+	if !ok {
+		return nil, ErrNoObject
+	}
+	return slices.Clone(d), nil
+}
+
+func (m *memObjects) Delete(_ context.Context, name string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.objs, name)
+	return nil
+}
+
+func (m *memObjects) Names(context.Context) ([]string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return slices.Sorted(maps.Keys(m.objs)), nil
+}
+
+func (m *memObjects) Fill(context.Context) (int64, int64, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.used(), m.capacity, nil
+}
+
+var errObjectsFullForTest = &objectsFullError{}
+
+type objectsFullError struct{}
+
+func (*objectsFullError) Error() string { return "maximum bytes exceeded" }
@@ -884,7 +884,9 @@ func (ix *Indexer) dry() *dryRun {
 		// requirements, findings, seat definitions, holds and their bounds are not overlaid, nothing is said, and no hold is
 		// weighed (Holds and Counted nil hold nothing); ReadFile only reads.
 		Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil,
-		Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: nil, OwnBuildState: nil, underSeats: true}
+		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}
 	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"} {
+			"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions"} {
 			named[f] = true
 		}
 		ty := reflect.TypeOf(Indexer{})
@@ -311,6 +311,27 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { //
 			t.Errorf("entry %s: digest %s against %s before; only mesh-tools' moves", ref.Module, ref.Digest, rev.Entries[i].Digest)
 		}
 	}
+	// W4.2's store: both revisions kept in catalog-revisions hold one object per entry once, the moved entry twice, and
+	// each list; a revision read back is its entries whole; an entry the bucket lost is an error, never left out.
+	kept := newMemObjects(0)
+	_, nextObjects, _ := CatalogRevisionOf(next, moved)
+	if err := KeepRevision(context.Background(), kept, rev, objects); err != nil {
+		t.Fatal(err)
+	}
+	if err := KeepRevision(context.Background(), kept, next, nextObjects); err != nil {
+		t.Fatal(err)
+	}
+	if names, _ := kept.Names(context.Background()); len(names) != len(mods.vals)+1+2 || kept.puts != len(names) {
+		t.Errorf("catalog-revisions holds %v after %d puts, want each entry once, the moved one twice, and two lists", names, kept.puts)
+	}
+	if back, got, err := ReadRevision(context.Background(), kept, "trunk.2"); err != nil || back.Source != 8 || len(got) != len(mods.vals) ||
+		!slices.ContainsFunc(got, func(e meshcatalog.CatalogEntry) bool { return e.Module == "mesh-tools" && e.Fp == tools.Fp }) {
+		t.Errorf("trunk.2 read back = %+v, %d entries, %v", back, len(got), err)
+	}
+	_ = kept.Delete(context.Background(), EntryObject(next.Entries[0].Digest))
+	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)
+	}
 	graph := ix.Graph.(*memStore)
 	written := graph.writes
 	if err := ix.Recompute(context.Background()); err != nil || graph.writes != written {
@@ -57,7 +57,11 @@ type manifest struct {
 	Slug    string   `json:"slug"`
 	RunsAs  string   `json:"runs-as"`
 	Streams []stream `json:"streams"`
-	Claims  []struct {
+	Objects []struct {
+		Name   string `json:"name"`
+		MaxMiB int    `json:"max-mib"`
+	} `json:"objects"`
+	Claims []struct {
 		Name   string   `json:"name"`
 		Scope  string   `json:"scope"`
 		Serves []string `json:"serves"`
@@ -119,6 +123,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 		{Name: index.HoldsState, History: 5}, {Name: index.EngineReadsState, History: 1}, {Name: index.HoldBoundsState, History: 1, KeyTTL: true}, {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17, ADR 0632 §2
 		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,
+	// capped at 512 MiB, under the local name the indexer opens.
+	if len(m.Objects) != 1 || m.Objects[0].Name != index.CatalogRevisionsObjects || m.Objects[0].MaxMiB != 512 {
+		t.Errorf("objects = %+v, want %s capped at 512 MiB", m.Objects, index.CatalogRevisionsObjects)
+	}
 	// D-11 (ADR 0634 §1): the holders and buckets the indexer names (internal/index/names.go) are its manifest's: its own
 	// name and states with their revisions, and the states it reads, each as `<holder>.<state>`.
 	if index.Module != m.Module {
@@ -99,6 +99,9 @@
    { "name": "pending", "history": 1 },
    { "name": "unjudged", "history": 1 }
  ],
+  "objects": [
+    { "name": "catalog-revisions", "max-mib": 512 }
+  ],
  "emits": [
    "take-in.pending",
    "take-in.taken",