Merge pull request 'Indexer: index a module registered from its own repository, so the grants part composes its grants (issue 1945)' (#63) from fix/indexer-registered-modules into main

This commit was merged in pull request #63.
This commit is contained in:
2026-10-26 11:33:41 +00:00
10 changed files with 456 additions and 6 deletions
@@ -266,11 +266,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}
errs := make(chan error, 1)
var buildsWatch jetstream.KeyWatcher
if builds != nil {
if w, err := settle(ctx, builds, each); err != nil {
process.NotWatched(logger.Printf, delivery.BuildsBucket, err, "a module whose key names no directory is not given one until the indexer starts again")
} else {
go func() { errs <- follow(ctx, w, each) }()
buildsWatch = w
}
}
@@ -285,7 +286,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), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call)}
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(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.
@@ -336,6 +337,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := ix.RecomputeAtStart(ctx); err != nil {
logger.Printf("computing the new-form fingerprints at start: %v", err)
}
// A module registered in the catalogue from its own repository gets its manifest indexed when the controller's records
// name it and the bucket holds none (registered.go); the records that follow are heard below.
if err := ix.IndexRegisteredAll(ctx); err != nil {
logger.Printf("indexing the registered modules at start: %v", err)
}
if buildsWatch != nil {
go func() {
errs <- follow(ctx, buildsWatch, func(e jetstream.KeyValueEntry) {
each(e)
if !opOf(e) {
return
}
var r delivery.BuildRecord
if json.Unmarshal(e.Value(), &r) != nil || r.Module == "" {
return
}
if err := ix.IndexRegisteredModule(ctx, r.Module); err != nil {
logger.Printf("the registered module %s is not indexed: %v", r.Module, 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)
@@ -2,11 +2,16 @@ package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"path"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-indexer/internal/index"
)
// commitTreesTool is the forge's tool for a commit's trees (mesh-catalog modules/gitea): with `paths`, by module directory
@@ -93,3 +98,75 @@ func containsVia(call toolCaller) func(ctx context.Context, repository, commit,
return *a.Contains, nil
}
}
// getFileTool is the forge's tool for one file at a ref (mesh-catalog modules/gitea gitea_get_file): arguments owner,
// repo, path and ref, answer {file: {content}}; a path it does not have fails with a 404 in its words.
const getFileTool = "gitea.gitea_get_file"
// lockFiles are read beside a manifest, as the forge bridge reads them (bridge.LockFiles).
var lockFiles = []string{"go.sum", "package-lock.json"}
// readVia reads a repository at a commit as the forge bridge does for a merged message: the trees by gitea_commit_trees,
// then the module.json and lock files of the one directory dir by gitea_get_file (a repository can hold hundreds). A manifest the
// forge does not have is an error; a directory with no module.json in the trees is answered with no manifest; a lock file it does not have is left out.
func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir string) (index.RepoRead, error) {
return func(_ context.Context, repository, sha, dir string) (index.RepoRead, error) {
owner, repo, ok := strings.Cut(delivery.RepositoryName(repository), "/")
if !ok {
return index.RepoRead{}, 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 index.RepoRead{}, err
}
var a struct {
Root string `json:"root"`
Trees map[string]string `json:"trees"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return index.RepoRead{}, fmt.Errorf("the answer of %s is not {root, trees}: %w", commitTreesTool, err)
}
out := index.RepoRead{Root: a.Root, Trees: a.Trees, Manifests: map[string]string{}, Locks: map[string]map[string]string{}}
file := func(dir, name string) (string, error) {
p := name
if dir != "" {
p = dir + "/" + name
}
raw, err := call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": p, "ref": sha})
if err != nil {
return "", err
}
var f struct {
File *struct {
Content *string `json:"content"`
} `json:"file"`
}
if err := json.Unmarshal(raw, &f); err != nil || f.File == nil || f.File.Content == nil {
return "", fmt.Errorf("the answer of %s is not {file: {content}}: %v", getFileTool, err)
}
return *f.File.Content, nil
}
if _, held := a.Trees[dir]; held {
m, err := file(dir, "module.json")
if err != nil {
return index.RepoRead{}, fmt.Errorf("%s: %w", path.Join(dir, "module.json"), err)
}
out.Manifests[dir] = m
for _, name := range lockFiles {
c, err := file(dir, name)
if err != nil {
if strings.Contains(err.Error(), ": 404 ") {
continue
}
return index.RepoRead{}, fmt.Errorf("%s: %w", path.Join(dir, name), err)
}
sum := sha256.Sum256([]byte(c))
if out.Locks[dir] == nil {
out.Locks[dir] = map[string]string{}
}
out.Locks[dir][name] = hex.EncodeToString(sum[:])
}
}
return out, nil
}
}
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"errors"
"slices"
"strings"
"testing"
@@ -50,3 +51,33 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
t.Error("the trunk asked for paths")
}
}
func TestReadViaReadsManifestsAndLocks(t *testing.T) {
var files []string
call := func(key string, body any) (json.RawMessage, error) {
switch key {
case commitTreesTool:
return json.RawMessage(`{"root":"r0","trees":{"":"t1","modules/other":"t2"}}`), nil
case getFileTool:
files = append(files, body.(map[string]string)["path"])
switch body.(map[string]string)["path"] {
case "module.json":
return json.RawMessage(`{"file":{"content":"{\"module\":\"snake\"}"}}`), nil
case "go.sum":
return json.RawMessage(`{"file":{"content":"x"}}`), nil
}
return nil, errors.New("Gitea API /x: 404 not found")
}
return nil, errors.New("unexpected " + key)
}
got, err := readVia(call)(context.Background(), "http://novox.internal:20000/jschoubben/snake.git", "abc", "")
if err != nil {
t.Fatal(err)
}
if len(got.Manifests) != 1 || len(files) != 3 { // the one directory asked: its manifest and two lock files, never the others
t.Fatalf("manifests = %v, files asked = %v", got.Manifests, files)
}
if got.Root != "r0" || got.Manifests[""] != `{"module":"snake"}` || len(got.Locks[""]) != 1 || got.Locks[""]["go.sum"] == "" {
t.Fatalf("read = %+v", got)
}
}
@@ -2,6 +2,8 @@ package index
import (
"encoding/json"
"maps"
"slices"
"strings"
"sync"
@@ -88,3 +90,50 @@ func (b *Builds) PathOf(module, repository string) (string, bool) {
}
return pathOf(strings.Trim(best.Path, "/")), true
}
// Registration is a module registered in the catalogue from its own repository, as the controller's newest outcome record
// of it names it: the repository, the directory of its manifest there ("" or "." for the root) and the commit it was
// built at.
type Registration struct {
Module, Repository, Path, Commit string
}
// registered is whether the record is an outcome of the trunk (no named branch, or main, master or trunk, as a repository's default branch is named) that did not fail: what registers a
// module. Unlike IsTrunkBuild it asks for no recorded source inputs, which no build records since ADR 0530 clause 1.
func registered(r delivery.BuildRecord) bool {
return r.State == "" && r.Failed == "" && trunkLike[r.Ref]
}
// RegistrationOf is the registration of a module by its newest registering record; false when there is none.
func (b *Builds) RegistrationOf(module string) (Registration, bool) {
var best *delivery.BuildRecord
b.mu.Lock()
defer b.mu.Unlock()
for _, r := range b.records {
if r.Module == module && registered(r) && (best == nil || newer(r, *best)) {
c := r
best = &c
}
}
if best == nil {
return Registration{}, false
}
return Registration{Module: best.Module, Repository: best.Repository, Path: best.Path, Commit: best.Commit}, true
}
// Registered is the registration of every module a record names, by module name.
func (b *Builds) Registered() []Registration {
b.mu.Lock()
names := map[string]bool{}
for _, r := range b.records {
names[r.Module] = true
}
b.mu.Unlock()
var out []Registration
for _, name := range slices.Sorted(maps.Keys(names)) {
if reg, ok := b.RegistrationOf(name); ok {
out = append(out, reg)
}
}
return out
}
@@ -53,6 +53,9 @@ type Indexer struct {
// 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)
// Read asks the forge for a repository's trees at a commit and the manifest and lock-file hashes of the one module directory dir (the
// root spelled ""), as the forge bridge reads them for a merged message (registered.go). Nil indexes no registered module.
Read func(ctx context.Context, repository, sha, dir string) (RepoRead, 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)
@@ -61,6 +64,7 @@ type Indexer struct {
source sourceNow
previews map[string]delivery.Preview // by previews key (previews.go)
toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go)
tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go)
recompute sync.Mutex // one Recompute at a time
}
@@ -25,6 +25,13 @@ import (
// A value the module already holds but for its commit is not written again: its commit stays the one its fingerprint first
// appeared at (writeOverlay), so a merge writes only the modules it changes.
func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
return ix.applyManifests(ctx, f, true)
}
// applyManifests is applyMerged; drop says whether the modules of the repository the message holds no manifest for are
// deleted. A message that names only some of a repository's modules (a registered module read on its own, registered.go)
// drops none.
func (ix *Indexer) applyManifests(ctx context.Context, f delivery.Forge, drop bool) error {
digests, err := ix.keepManifests(ctx, f)
if err != nil {
return err
@@ -84,7 +91,7 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
}
// 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 drop && 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
}
@@ -0,0 +1,129 @@
package index
import (
"context"
"fmt"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A module registered in the catalogue from its own repository (the operator registers it with the repository and the
// path of its manifest) had no manifest here: the indexer learns a repository only from a merged message, and no merge
// followed the registration. The controller's outcome records name each registered module's repository, directory and
// commit (builds.go Registered), so the indexer reads that module's manifest at the repository's head (the gate's source names it) or, with none named, at the
// registered commit, and writes the same keys a merge writes. A later merge to the repository's trunk reaches the module through the forge bridge's merged
// message as for every other repository.
// RepoRead is a repository at a commit as the forge bridge reads it for a merged message: the root tree, the tree of each
// directory holding a module.json (the root spelled ""), their manifests and the hex sha256 of the lock files beside them.
type RepoRead struct {
Root string
Trees map[string]string
Manifests map[string]string
Locks map[string]map[string]string
}
// IndexRegistered writes the module of a registration as a merge of its repository would, when the modules bucket holds
// none of that name. Its commit is the head the gate's source names for the repository, else the registration's. The
// repository's other modules are neither written nor dropped: they are the merges'. A manifest that is not at the path or
// names another module is said and left.
func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error {
if ix.Modules == nil || ix.Read == nil {
return nil
}
key, err := delivery.ModulesKey(r.Module)
if err != nil {
ix.logf("%s: the registered module has no usable key, skipped: %v", r.Module, err)
return nil
}
if _, _, found, err := ix.Modules.Get(ctx, key); err != nil {
return fmt.Errorf("modules.%s: %w", key, err)
} else if found {
return nil
}
repository := delivery.RepositoryName(r.Repository)
commit := r.Commit
dir := strings.Trim(r.Path, "/")
if dir == delivery.RootPath {
dir = ""
}
// The head the gate's source names for the repository, when it names one, else the registration's commit. A module a
// merge removed is absent at the head and is said and left.
ix.mu.Lock()
for repo, head := range ix.source.heads {
if delivery.SameRepository(repo, r.Repository) && head != "" {
commit = head
}
}
ix.mu.Unlock()
if repository == "" || commit == "" {
ix.logf("%s: registered with no repository or commit, so its manifest is not read", r.Module)
return nil
}
// One try for each repository, commit and directory: a repository with no manifest where it was registered is said once,
// not at every record of the controller's builds.
tried := repository + "@" + commit + ":" + dir
ix.mu.Lock()
if ix.tried[tried] {
ix.mu.Unlock()
return nil
}
if ix.tried == nil {
ix.tried = map[string]bool{}
}
ix.tried[tried] = true
ix.mu.Unlock()
read, err := ix.Read(ctx, repository, commit, dir)
if err != nil {
ix.mu.Lock()
delete(ix.tried, tried) // a forge that could not say is asked again
ix.mu.Unlock()
return fmt.Errorf("%s: reading %s at %s: %w", r.Module, repository, commit, err)
}
raw, ok := read.Manifests[dir]
if !ok || read.Trees[dir] == "" {
ix.logf("%s: %s at %s holds no module.json at %q, so the registered module is not indexed", r.Module, repository, commit, dir)
return nil
}
if name := manifestModule(raw); name != r.Module {
ix.logf("%s: the manifest at %q of %s at %s names the module %q, so the registered module is not indexed", r.Module, dir, repository, commit, name)
return nil
}
f := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: repository, Merge: commit, Root: read.Root,
Trees: map[string]string{dir: read.Trees[dir]},
Manifests: map[string]string{dir: raw}}
if l := read.Locks[dir]; len(l) > 0 {
f.Locks = map[string]map[string]string{dir: l}
}
return ix.applyManifests(ctx, f, false)
}
// IndexRegisteredModule indexes the registration the controller's records name for a module, when there is one.
func (ix *Indexer) IndexRegisteredModule(ctx context.Context, module string) error {
if ix.Builds == nil {
return nil
}
if r, ok := ix.Builds.RegistrationOf(module); ok {
return ix.IndexRegistered(ctx, r)
}
return nil
}
// IndexRegisteredAll indexes every registered module the modules bucket does not hold. A module that cannot be read is
// said and the others go on; the first error is answered.
func (ix *Indexer) IndexRegisteredAll(ctx context.Context) error {
if ix.Builds == nil {
return nil
}
var first error
for _, r := range ix.Builds.Registered() {
if err := ix.IndexRegistered(ctx, r); err != nil {
ix.logf("the registered module %s is not indexed: %v", r.Module, err)
if first == nil {
first = err
}
}
}
return first
}
@@ -0,0 +1,129 @@
package index
import (
"context"
"errors"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func TestARegisteredModuleIsIndexedFromItsRepository(t *testing.T) { // the grants part's gaps: modules registered from their own repositories had no manifest here
const forge = "http://novox.internal:20000/jschoubben/snake.git"
snake := `{"module":"snake","version":"1"}`
for _, tc := range []struct {
name string
path string
read RepoRead
err error
ref string // the build record's ref
heads map[string]string // the gate source's heads
wantRead bool // the forge is read
held bool // modules.snake exists already
other bool // modules.other of the same repository exists
want bool // modules.snake is written
}{
{wantRead: true, name: "a module at the root of its repository", path: "", want: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "a root spelled as a dot", path: ".", want: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "a module in a directory", path: "modules/snake", want: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"modules/snake": "t1"}, Manifests: map[string]string{"modules/snake": snake}}},
{wantRead: true, name: "another module of the repository is neither written nor dropped", path: "", want: true, other: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{name: "a module the bucket holds is left", path: "", held: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "no manifest at the path", path: "modules/x",
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "a manifest of another module", path: "",
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": `{"module":"other","version":"1"}`}}},
{wantRead: true, name: "a repository whose default branch is master", path: "", ref: "master", want: true,
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{name: "a record of another branch", path: "", ref: "feat/x",
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "a repository the source names, the module unheld and present at the head", path: "", want: true, heads: map[string]string{"jschoubben/snake": h2},
read: RepoRead{Root: "r0", Trees: map[string]string{"": "t1"}, Manifests: map[string]string{"": snake}}},
{wantRead: true, name: "a forge that cannot say", path: "", err: errors.New("down")},
} {
t.Run(tc.name, func(t *testing.T) {
ix, mods := modulesIndexer(t, rec(t, "build-1", "snake", forge, tc.path, tc.ref, "", "2026-10-19T09:00:00Z", ""))
if tc.held {
if err := mods.Create(context.Background(), "snake", mustModule(t, delivery.Module{Repository: "jschoubben/snake", Commit: h1, Path: ".", None2: "x"})); err != nil {
t.Fatal(err)
}
}
if tc.other {
if err := mods.Create(context.Background(), "other", mustModule(t, delivery.Module{Repository: "jschoubben/snake", Commit: h1, Path: "modules/other", None2: "x"})); err != nil {
t.Fatal(err)
}
}
ix.Manifests = newMem()
var asked struct {
repo, sha, dir string
n int
}
ix.Read = func(_ context.Context, repo, sha, dir string) (RepoRead, error) {
asked.repo, asked.sha, asked.dir = repo, sha, dir
asked.n++
return tc.read, tc.err
}
if tc.heads != nil {
ix.source = sourceNow{rev: 1, heads: tc.heads}
}
err := ix.IndexRegisteredAll(context.Background())
if (err != nil) != (tc.err != nil) {
t.Fatalf("error = %v", err)
}
wantCommit := "7a74cee83f88e6585a5511c3cc53bdf0f5303d20" // the record's
if tc.heads != nil {
wantCommit = h2 // the source's head
}
if tc.wantRead && asked.sha != wantCommit {
t.Errorf("the forge was asked at %s, want %s", asked.sha, wantCommit)
}
wantReads := 1
if !tc.wantRead {
wantReads = 0
}
if tc.wantRead && asked.dir != dirOf(tc.path) {
t.Errorf("the forge was asked for directory %q", asked.dir)
}
if asked.n != wantReads {
t.Errorf("forge reads = %d, want %d", asked.n, wantReads)
}
m, ok := moduleOf(t, mods, "snake")
if tc.want {
if !ok || m.Manifest == "" || m.Repository != "jschoubben/snake" || m.Commit != wantCommit || m.Path != pathOf(dirOf(tc.path)) {
t.Fatalf("modules.snake = %+v, found %v", m, ok)
}
if len(ix.Manifests.(*memStore).vals) != 1 {
t.Errorf("manifests written = %d, want 1", len(ix.Manifests.(*memStore).vals))
}
if asked.repo != "jschoubben/snake" {
t.Errorf("the forge was asked for %q", asked.repo)
}
} else if !tc.held && ok {
t.Fatalf("modules.snake written: %+v", m)
}
if _, ok := moduleOf(t, mods, "other"); ok != tc.other {
t.Errorf("modules.other held = %v, want %v", ok, tc.other)
}
})
}
}
func mustModule(t *testing.T, m delivery.Module) []byte {
t.Helper()
b, err := delivery.EncodeModule(withV(m))
if err != nil {
t.Fatal(err)
}
return b
}
func dirOf(p string) string {
if p == "." {
return ""
}
return p
}
@@ -95,10 +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)
}
// Two invokes: the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
// Three invokes: the file of a registered module's manifest and lock files (registered.go), 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; and the module directories of the trunk, issue 1900) and whether a
// commit is in another (issue 1900).
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file"}; !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).
+2 -1
View File
@@ -23,7 +23,8 @@
],
"invokes": [
"gitea.gitea_commit_trees",
"gitea.gitea_contains"
"gitea.gitea_contains",
"gitea.gitea_get_file"
],
"own-secrets": {
"broker": "${dir:state}/broker"