Drop the indexer's modules keys the trunk no longer holds, at each merge and at start (issue 1900)

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -295,7 +295,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	defer calls.Close()
 	ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf,
-		Named: namedVia(calls.Call)}
+		Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call)}
 	// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
 	// main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does
 	// not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate.
@@ -333,6 +333,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err := ix.RecomputeForBuild(ctx); err != nil {
 		logger.Printf("computing the new-form fingerprints at start: %v", err)
 	}
+	// The modules the trunk no longer holds go from the bucket (issue 1900); a merge does the same from its own message.
+	if err := ix.Reconcile(ctx); err != nil {
+		logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
+	}
 	// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
 	// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
 	// with the build module, as with the gate's source.

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go

@@ -46,3 +46,29 @@ func namedVia(call toolCaller) func(ctx context.Context, repository, sha string,
 		return a.Paths, nil
 	}
 }
+
+// trunkVia asks the forge for the module directories of a commit (issue 1900): the keys of the `trees` the same tool
+// answers without paths, one per directory that holds a module.json (the root as ""; the tool lists no other).
+func trunkVia(call toolCaller) func(ctx context.Context, repository, sha string) ([]string, error) {
+	return func(_ context.Context, repository, sha string) ([]string, error) {
+		owner, repo, ok := strings.Cut(delivery.RepositoryName(repository), "/")
+		if !ok {
+			return nil, fmt.Errorf("%q is not <owner>/<name>", repository)
+		}
+		raw, err := call(commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha})
+		if err != nil {
+			return nil, err
+		}
+		var a struct {
+			Trees map[string]string `json:"trees"`
+		}
+		if err := json.Unmarshal(raw, &a); err != nil {
+			return nil, fmt.Errorf("the answer of %s is not {root, trees}: %w", commitTreesTool, err)
+		}
+		dirs := make([]string, 0, len(a.Trees))
+		for dir := range a.Trees {
+			dirs = append(dirs, dir)
+		}
+		return dirs, nil
+	}
+}

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go

@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"slices"
+	"strings"
 	"testing"
 )
 
@@ -29,3 +31,22 @@ func TestNamedViaAsksTheTreesToolWithThePathsAndTakesTheirObjects(t *testing.T)
 		t.Error("an answer missing a path asked was taken")
 	}
 }
+
+func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
+	var gotArgs any
+	call := func(key string, body any) (json.RawMessage, error) {
+		gotArgs = body
+		return json.RawMessage(`{"root":"r","trees":{"modules/a":"x","modules/b":"y"}}`), nil
+	}
+	dirs, err := trunkVia(call)(context.Background(), "novox/mesh-catalog", strings.Repeat("a", 40))
+	if err != nil {
+		t.Fatal(err)
+	}
+	slices.Sort(dirs)
+	if !slices.Equal(dirs, []string{"modules/a", "modules/b"}) {
+		t.Errorf("dirs = %v", dirs)
+	}
+	if _, has := gotArgs.(map[string]any)["paths"]; has {
+		t.Error("the trunk asked for paths")
+	}
+}

modules/mesh-delivery-indexer/internal/index/candidates.go

@@ -52,6 +52,7 @@ type Indexer struct {
 	// manifest's paths (ADR 0495 clause 1, the lead's decision L8 of R4e). Nil asks nothing, and a module naming paths
 	// has none.
 	Named func(ctx context.Context, repository, sha string, paths map[string][]string) (map[string]map[string]string, error)
+	Trunk func(ctx context.Context, repository, sha string) ([]string, error)
 	Logf  func(format string, args ...any)
 
 	mu         sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach

modules/mesh-delivery-indexer/internal/index/inputs_test.go

@@ -111,7 +111,8 @@ func TestABaseEntersByTheFingerprintOfItsModuleOrNamesItsAbsence(t *testing.T) {
 	}
 	// Through the indexer: modules.mesh-tools holds a fingerprint of the new form (a record written by a later release).
 	ix, mods, _ := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
-	mods.put("mesh-tools", mustEncode(t, delivery.Module{Repository: catalogueRepo, Commit: merge, None: "x", FP2: fp, Manifest: strings.Repeat("a", 64)}))
+	mods.put("mesh-tools", mustEncode(t, delivery.Module{Repository: "novox/mesh-sdk", // a base held of another repository stays when the catalogue merges (issue 1900)
+		Commit: merge, None: "x", FP2: fp, Manifest: strings.Repeat("a", 64)}))
 	f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "t"}), map[string]string{"modules/avahi": aModuleWithABase(t)}, nil)
 	if err := ix.Apply(context.Background(), f); err != nil {
 		t.Fatal(err)

modules/mesh-delivery-indexer/internal/index/modules.go

@@ -2,8 +2,10 @@ package index
 
 import (
 	"context"
+	"fmt"
 	"maps"
 	"slices"
+	"strings"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -33,18 +35,22 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
 	objects, namedErr := ix.namedObjects(ctx, f)
 	overlay := map[string]delivery.Module{}
 	dirOf := map[string]string{}
+	named, every := map[string]bool{}, true // every module the message's manifests name, and whether each names one
 	for _, dir := range slices.Sorted(maps.Keys(f.Manifests)) {
 		raw := f.Manifests[dir]
 		name := manifestModule(raw)
 		if name == "" {
 			ix.logf("%s at %s: the manifest in %q names no module, so no module is written for it", f.Repository, f.Merge, dir)
+			every = false
 			continue
 		}
 		key, err := delivery.ModulesKey(name)
 		if err != nil {
+			every = false
 			ix.logf("%s: the module %q has no usable key, skipped: %v", f.Repository, name, err)
 			continue
 		}
+		named[key] = true
 		cur, has := held[key]
 		if has && cur.Repository != "" && !delivery.SameRepository(cur.Repository, f.Repository) {
 			ix.logf("%s at %s: %q names the module %s, which the indexer holds of %s; left as it is (a move between repositories is the operator's)", f.Repository, f.Merge, dir, name, cur.Repository)
@@ -70,6 +76,13 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
 		}
 		overlay[key] = next
 	}
+	// A module of this repository the merge commit holds no module.json for is gone from the trunk, and so is its key
+	// (issue 1900). A message that carries no manifests (an older writer) or one that cannot be read whole says nothing.
+	if len(f.Manifests) > 0 && every {
+		if err := ix.dropRemoved(ctx, f.Repository, held, func(key string, _ delivery.Module) bool { return named[key] }); err != nil {
+			return err
+		}
+	}
 	// What this merge moved reaches the modules standing on it (a base's tree, build-agent's fingerprint), in the graph's order.
 	return ix.recomputeWith(ctx, overlay)
 }
@@ -149,3 +162,74 @@ func (ix *Indexer) namedObjects(ctx context.Context, f delivery.Forge) (map[stri
 	}
 	return objects, nil
 }
+
+// dropRemoved deletes, from the modules the bucket holds of the repository, each one the trunk no longer holds (issue 1900):
+// held(key, module) says whether the trunk holds it. A module of another repository, or of none known, is never touched.
+func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[string]delivery.Module, trunkHolds func(key string, m delivery.Module) bool) error {
+	for _, key := range slices.Sorted(maps.Keys(held)) {
+		cur := held[key]
+		if cur.Repository == "" || !delivery.SameRepository(cur.Repository, repository) || trunkHolds(key, cur) {
+			continue
+		}
+		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
+			if !found {
+				return nil, false, nil
+			}
+			if m, err := delivery.DecodeModule(value); err != nil || !delivery.SameRepository(m.Repository, repository) {
+				return nil, false, nil
+			}
+			return nil, true, nil
+		})
+		if err != nil {
+			return fmt.Errorf("modules.%s: %w", key, err)
+		}
+		ix.logf("modules.%s deleted: %s holds no module of that name at its trunk (issue 1900)", key, repository)
+	}
+	return nil
+}
+
+// Reconcile deletes the modules keys whose module the trunk does not hold, for each repository the latest source names a
+// head of: the module directories at that head are asked of the forge (Trunk), and a module of the repository whose
+// directory is not among them goes (issue 1900). It is run at the indexer's start; a merge does the same from its own
+// message (applyMerged). A repository whose trunk cannot be read, or reads as holding no module at all, is left as it is
+// and said: a failure never empties the bucket.
+func (ix *Indexer) Reconcile(ctx context.Context) error {
+	if ix.Modules == nil || ix.Trunk == nil {
+		return nil
+	}
+	ix.mu.Lock()
+	heads := maps.Clone(ix.source.heads)
+	ix.mu.Unlock()
+	held, err := ix.heldModules(ctx)
+	if err != nil {
+		return err
+	}
+	for _, repository := range slices.Sorted(maps.Keys(heads)) {
+		mine := map[string]delivery.Module{}
+		for key, m := range held {
+			if m.Repository != "" && delivery.SameRepository(m.Repository, repository) {
+				mine[key] = m
+			}
+		}
+		if len(mine) == 0 {
+			continue
+		}
+		dirs, err := ix.Trunk(ctx, repository, heads[repository])
+		if err != nil {
+			ix.logf("%s at %s: the modules of the trunk cannot be read, so none is dropped: %v", repository, heads[repository], err)
+			continue
+		}
+		if len(dirs) == 0 {
+			ix.logf("%s at %s: the trunk reads as holding no module, so none is dropped", repository, heads[repository])
+			continue
+		}
+		there := map[string]bool{}
+		for _, d := range dirs {
+			there[pathOf(strings.Trim(d, "/"))] = true
+		}
+		if err := ix.dropRemoved(ctx, repository, mine, func(_ string, m delivery.Module) bool { return there[m.Path] }); err != nil {
+			return err
+		}
+	}
+	return nil
+}

modules/mesh-delivery-indexer/internal/index/remove_test.go

@@ -0,0 +1,117 @@
+package index
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+func twoModules(t *testing.T) (*Indexer, *memStore) {
+	t.Helper()
+	ix, mods := modulesIndexer(t)
+	if err := ix.Apply(context.Background(), mergedMsg(catalogueRepo, map[string]string{"modules/a": "t", "modules/b": "t"})); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"a", "b"} {
+		if _, ok := mods.vals[k]; !ok {
+			t.Fatalf("setup: modules.%s not written", k)
+		}
+	}
+	return ix, mods
+}
+
+func TestAMergeThatRemovesAModuleDeletesItsKey(t *testing.T) { // 1900/AC1
+	ix, mods := twoModules(t)
+	if err := ix.Apply(context.Background(), mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := mods.vals["b"]; ok {
+		t.Error("modules.b is still held after a merge whose trunk has no module b")
+	}
+	if _, ok := mods.vals["a"]; !ok {
+		t.Error("modules.a, which the trunk still holds, was deleted")
+	}
+}
+
+func TestAMergeLeavesTheModulesOfAnotherRepository(t *testing.T) { // 1900/AC1 (scope)
+	ix, mods := twoModules(t)
+	if err := ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/c": "t"})); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := mods.vals["a"]; !ok {
+		t.Error("a merge of another repository deleted modules.a")
+	}
+	if _, ok := mods.vals["b"]; !ok {
+		t.Error("a merge of another repository deleted modules.b")
+	}
+}
+
+func TestAMergeWithoutManifestsDeletesNothing(t *testing.T) { // 1900/AC1: older writers send none, which does not say the trunk is empty
+	ix, mods := twoModules(t)
+	f := mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})
+	f.Manifests = nil
+	if err := ix.Apply(context.Background(), f); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := mods.vals["b"]; !ok {
+		t.Error("a merge message with no manifests deleted modules.b")
+	}
+}
+
+func TestAPathWithoutAModuleJSONIsNeverAModule(t *testing.T) { // 1900/AC2: testdata such as fixture-build
+	ix, mods := twoModules(t)
+	f := mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})
+	f.Trees["testdata/fixture-build"] = "t" // a directory in the tree whose module.json the message does not carry
+	if err := ix.Apply(context.Background(), f); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := mods.vals["fixture-build"]; ok {
+		t.Error("modules.fixture-build written for a path without a module.json")
+	}
+}
+
+func TestStartingDeletesTheKeysTheTrunkDoesNotHold(t *testing.T) { // 1900/AC3
+	ix, mods := twoModules(t)
+	ctx := context.Background()
+	if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil {
+		t.Fatal(err)
+	}
+	var asked string
+	ix.Trunk = func(_ context.Context, repository, sha string) ([]string, error) {
+		asked = repository + "@" + sha
+		return []string{"modules/a"}, nil
+	}
+	if err := ix.Reconcile(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if asked != catalogueRepo+"@"+merge {
+		t.Errorf("the trunk was asked as %q, want the source's head of the repository", asked)
+	}
+	if _, ok := mods.vals["b"]; ok {
+		t.Error("modules.b is still held though the trunk holds no module at modules/b")
+	}
+	if _, ok := mods.vals["a"]; !ok {
+		t.Error("modules.a was deleted")
+	}
+}
+
+func TestStartingDeletesNothingWhenTheTrunkCannotBeRead(t *testing.T) { // 1900/AC3: failure is loud, never a mass delete
+	ix, mods := twoModules(t)
+	ctx := context.Background()
+	if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil {
+		t.Fatal(err)
+	}
+	for name, trunk := range map[string]func(context.Context, string, string) ([]string, error){
+		"an error":         func(context.Context, string, string) ([]string, error) { return nil, errors.New("forge down") },
+		"no module at all": func(context.Context, string, string) ([]string, error) { return nil, nil },
+	} {
+		ix.Trunk = trunk
+		_ = ix.Reconcile(ctx)
+		if len(mods.vals) != 2 {
+			t.Errorf("%s: %d keys held, want 2 untouched", name, len(mods.vals))
+		}
+	}
+	_ = delivery.RootPath
+}