Merge pull request 'Drop the indexer's modules keys the catalogue's trunk no longer holds, at each merge and at start (issue 1900)' (#26) from fix/1900-indexer-drops-removed-modules into main

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), Contains: containsVia(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,50 @@ 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
+	}
+}
+
+// containsVia asks the forge whether a commit is in another (gitea.gitea_contains: contains is true for the same commit or an ancestor).
+func containsVia(call toolCaller) func(ctx context.Context, repository, commit, in string) (bool, error) {
+	return func(_ context.Context, repository, commit, in string) (bool, error) {
+		owner, repo, ok := strings.Cut(delivery.RepositoryName(repository), "/")
+		if !ok {
+			return false, fmt.Errorf("%q is not <owner>/<name>", repository)
+		}
+		raw, err := call("gitea.gitea_contains", map[string]any{"owner": owner, "repo": repo, "commit": commit, "in": in})
+		if err != nil {
+			return false, err
+		}
+		var a struct {
+			Contains *bool `json:"contains"`
+		}
+		if err := json.Unmarshal(raw, &a); err != nil || a.Contains == nil {
+			return false, fmt.Errorf("the answer of gitea.gitea_contains names no contains: %v", err)
+		}
+		return *a.Contains, 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,7 +52,10 @@ 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)
-	Logf  func(format string, args ...any)
+	Trunk func(ctx context.Context, repository, sha string) ([]string, error)
+	// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains).
+	Contains func(ctx context.Context, repository, commit, in string) (bool, error)
+	Logf     func(format string, args ...any)
 
 	mu         sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
 	source     sourceNow

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-tools", // 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,94 @@ 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
+		}
+		head := heads[repository]
+		// The source can lag a merge already applied: a key written at a commit the head read does not hold (a merge heard
+		// first) is not read as removed, only one whose commit the head is, or contains.
+		stays := func(_ string, m delivery.Module) bool {
+			if m.Path == "" || there[m.Path] {
+				return true
+			}
+			if m.Commit == head {
+				return false
+			}
+			if m.Commit == "" || ix.Contains == nil {
+				return true
+			}
+			in, err := ix.Contains(ctx, repository, m.Commit, head)
+			if err != nil {
+				ix.logf("%s: whether %s is in %s cannot be read, so the module at %s is kept: %v", repository, m.Commit, head, m.Path, err)
+				return true
+			}
+			return !in
+		}
+		if err := ix.dropRemoved(ctx, repository, mine, stays); err != nil {
+			return err
+		}
+	}
+	return nil
+}

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

@@ -0,0 +1,144 @@
+package index
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"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
+	}
+	ix.Contains = func(context.Context, string, string, string) (bool, error) { return true, 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
+}
+
+func TestAKeyWrittenAheadOfTheSourceHeadStaysAndOneWithoutAPathIsSkipped(t *testing.T) { // 1900/AC3 (review): the source lags a merge already applied
+	ix, mods := twoModules(t) // written at the merge commit C
+	ctx := context.Background()
+	const older = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+	if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, older)); err != nil {
+		t.Fatal(err)
+	}
+	mods.put("nopath", mustEncode(t, delivery.Module{Repository: catalogueRepo, Commit: older, None2: "x", Manifest: strings.Repeat("a", 64)}))
+	ix.Trunk = func(context.Context, string, string) ([]string, error) { return []string{"modules/a"}, nil }
+	ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == in, nil } // C is not in B
+	if err := ix.Reconcile(ctx); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"a", "b", "nopath"} {
+		if _, ok := mods.vals[k]; !ok {
+			t.Errorf("modules.%s was deleted", k)
+		}
+	}
+	ix.Contains = func(context.Context, string, string, string) (bool, error) { return false, errors.New("forge down") }
+	_ = ix.Reconcile(ctx)
+	if _, ok := mods.vals["b"]; !ok {
+		t.Error("modules.b deleted though whether its commit is in the head cannot be read")
+	}
+}

modules/mesh-delivery-indexer/manifest_test.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"regexp"
 	"slices"
 	"strings"
 	"testing"
@@ -94,9 +95,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 	if m.SecretsOwner != "mesh-delivery-indexer" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	// One invoke: the forge's trees with the named paths' objects, since the indexer is the one reader of a manifest's paths
+	// Two invokes: the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
-	// (ADR 0499 clauses 2 and 3, ADR 0495 clause 1, the lead's decision L8 of R4e).
+	// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
-	if want := []string{"gitea.gitea_commit_trees"}; !reflect.DeepEqual(m.Invokes, want) {
+	// commit is in another (issue 1900).
+	if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains"}; !reflect.DeepEqual(m.Invokes, want) {
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)
 	}
 	// No emits, no streams, no bus, no seat, no tool (ADR 0437 §4).
@@ -169,3 +171,32 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR
 		t.Fatalf("read %d source files, want the program's", files)
 	}
 }
+
+// Every forge tool the program calls is granted in invokes, or the bus refuses the call (issue 1900).
+func TestEveryGiteaToolTheCodeCallsIsInInvokes(t *testing.T) {
+	m, _ := load(t)
+	files, err := filepath.Glob("cmd/mesh-delivery-indexer/*.go")
+	if err != nil || len(files) == 0 {
+		t.Fatalf("no program files: %v", err)
+	}
+	re := regexp.MustCompile(`gitea\.gitea_[a-z_]+`)
+	seen := 0
+	for _, f := range files {
+		if strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		src, err := os.ReadFile(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, tool := range re.FindAllString(string(src), -1) {
+			seen++
+			if !slices.Contains(m.Invokes, tool) {
+				t.Errorf("%s calls %s, which invokes does not grant", f, tool)
+			}
+		}
+	}
+	if seen < 2 {
+		t.Errorf("found %d tool names in the program, want at least the two it calls", seen)
+	}
+}

modules/mesh-delivery-indexer/module.json

@@ -21,7 +21,8 @@
     "mesh-delivery-gate.source"
   ],
   "invokes": [
-    "gitea.gitea_commit_trees"
+    "gitea.gitea_commit_trees",
+    "gitea.gitea_contains"
   ],
   "own-secrets": {
     "broker": "${dir:state}/broker"