From 80bc3a408464436822307b7e08acc6fdee53abbd Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 20:56:35 +0100 Subject: [PATCH] The indexer runs on its view of the trunk read from the newest catalog graph revision, keeps it after every write and takes the trunk in at a first start (milestone 47/G W4.3, the view wired) main.go reads the view and the module graph's nodes (NodeOf) from the newest trunk revision instead of the modules and graph states; every write outside a recompute (the judge pass's verdicts, a build's imports, the source stamp, the modules the trunk dropped, the indexer's own entry a held revision writes) ends in KeepView, so a restart loses nothing; a revision lists every module the view holds, one whose manifest cannot be read too. At a first start whose objects keep no trunk revision the indexer takes in the head the gate's source names for every repository by the forge bridge's merged message and keeps revision 1 (the lead's ruling A). The whole-module tests read the view from the revisions. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../mesh-delivery-indexer/integration_test.go | 103 ++++++++++++++--- .../cmd/mesh-delivery-indexer/main.go | 31 +++-- .../internal/index/candidates.go | 15 ++- .../internal/index/catalog_graph.go | 22 ++++ .../internal/index/catalog_revisions.go | 109 ++++++++++++++---- .../internal/index/imports.go | 7 ++ .../internal/index/modules.go | 7 ++ .../internal/index/pending.go | 5 + .../internal/index/source.go | 90 +++++++++++++++ .../internal/index/toolchains_test.go | 77 ++++++++++++- .../internal/index/verdicts.go | 4 + 11 files changed, 417 insertions(+), 53 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 2d127854..efafb1e5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -39,6 +39,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" @@ -86,7 +87,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 }, func(ctx context.Context, js jetstream.JetStream) { // The states and the stream the controller makes from the indexer's manifest (old sections, ADR 0595 §11). for bucket, history := range map[string]uint8{index.PendingBucket: 1, index.VerdictsBucket: index.VerdictsHistory, - index.GraphBucket: 1, index.TakenBucket: 1, index.PackagesBucket: 1, + index.TakenBucket: 1, index.PackagesBucket: 1, index.SeatDefinitionsBucket: 1, index.RequirementsBucket: 1, findingsBucket: 1, index.HoldsBucket: 1, index.UnjudgedBucket: 1} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: history}); err != nil { t.Fatal(err) @@ -262,6 +263,13 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 named := held.WaitsFor get := func(bucket, key string) ([]byte, bool) { t.Helper() + if bucket == index.ModulesBucket { + e, err := revisionView{js: js}.Get(ctx, key) + if err != nil { + return nil, false + } + return e.Value(), true + } kv, err := js.KeyValue(ctx, bucket) if err != nil { t.Fatal(err) @@ -284,9 +292,6 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 if _, ok := get(index.VerdictsBucket, own.Manifest); !ok { t.Errorf("the verdict on the indexer's manifest %s is not written", own.Manifest) } - if gk, _ := graph.Key(ix); !func() bool { _, ok := get(index.GraphBucket, gk); return ok }() { - t.Errorf("the indexer's key %s of the module graph is not written", gk) - } repoKey, _ := trunk.RepositoryKey(m1.Repository) if _, ok := get(index.ModulesBucket, "a"); ok { t.Errorf("modules.a is written while the revision is pending") @@ -471,7 +476,75 @@ var acceptedHeard atomic.Int64 // startIndexerRun starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill // mesh-controller_builds, and runs the program; restart starts it again. mesh-delivery-gate_source is never created. -func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue, func()) { +// revisionView reads the indexer's view of the trunk as the indexer itself does at a restart (milestone 47/G): the newest +// trunk revision its objects catalog-revisions keep, a module by its key. +type revisionView struct{ js jetstream.JetStream } + +func (v revisionView) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { + view, _, _, err := index.TrunkView(ctx, &busObjects{js: v.js, local: index.CatalogRevisionsObjects}) + if err != nil { + return nil, err + } + value, rev, found, err := view.Get(ctx, key) + if err != nil { + return nil, err + } + if !found { + return nil, jetstream.ErrKeyNotFound + } + return viewEntry{key: key, value: value, revision: rev}, nil +} + +// History is the module's value in every trunk revision the objects keep that holds it, oldest first. +func (v revisionView) History(ctx context.Context, key string) ([]jetstream.KeyValueEntry, error) { + o := &busObjects{js: v.js, local: index.CatalogRevisionsObjects} + names, err := o.Names(ctx) + if err != nil { + return nil, err + } + var revs []int64 + for _, n := range names { + if r, ok := index.TrunkRevisionOf(n); ok { + revs = append(revs, r) + } + } + slices.Sort(revs) + var out []jetstream.KeyValueEntry + for _, r := range revs { + _, entries, err := index.ReadRevision(ctx, o, fmt.Sprintf("trunk.%d", r)) + if err != nil { + return nil, err + } + for _, e := range entries { + if e.Module != key { + continue + } + data, err := meshcatalogclient.EncodeModule(index.ModuleValueOf(e)) + if err != nil { + return nil, err + } + out = append(out, viewEntry{key: key, value: data, revision: uint64(r)}) + } + } + return out, nil +} + +// viewEntry is one module's value of the view. +type viewEntry struct { + key string + value []byte + revision uint64 +} + +func (e viewEntry) Bucket() string { return index.CatalogRevisionsObjects } +func (e viewEntry) Key() string { return e.key } +func (e viewEntry) Value() []byte { return e.value } +func (e viewEntry) Revision() uint64 { return e.revision } +func (e viewEntry) Created() time.Time { return time.Time{} } +func (e viewEntry) Delta() uint64 { return 0 } +func (e viewEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } + +func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, revisionView, func()) { dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") if err != nil { t.Fatal(err) @@ -526,10 +599,11 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . if err != nil { t.Fatal(err) } - modules, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.ModulesBucket, History: index.ModulesHistory}) - if err != nil { + // The indexer's objects catalog-revisions, which the grants part creates from its manifest's old objects section. + if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.ObjectsBucket(indexerModule, index.CatalogRevisionsObjects)}); err != nil { t.Fatal(err) } + modules := revisionView{js: js} // The bucket of the manifests, made by the controller from the manifest's state entry manifests (history 1, issue 1449). if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.ManifestsBucket, History: index.ManifestsHistory}); err != nil { t.Fatal(err) @@ -583,7 +657,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . } // startIndexer is startIndexerRun for a test that never starts the program again. -func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) { +func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, revisionView) { js, ctx, candidates, modules, _ := startIndexerRun(t, put, before...) return js, ctx, candidates, modules } @@ -960,17 +1034,16 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC if _, err := manifests.Put(ctx, mkey, mdata); err != nil { t.Fatal(err) } - stale, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-catalog", Commit: "89abcdef0123456789abcdef0123456789abcdef", Path: "modules/avahi", + // The view the indexer starts from is the newest trunk revision its objects keep (milestone 47/G). + stale := meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-catalog", Commit: "89abcdef0123456789abcdef0123456789abcdef", Path: "modules/avahi", Manifest: digest, None2: "the manifest cannot be read: claims node-build-go, which is a seat in the mesh's own namespace that it does not define", - Inputs2: &meshcatalog.ModulesByModuleValueInputs2{Tree: "tree-avahi", Locks: map[string]string{"go.sum": goSum}}}) + Inputs2: &meshcatalog.ModulesByModuleValueInputs2{Tree: "tree-avahi", Locks: map[string]string{"go.sum": goSum}}} + rev, objects, err := index.CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 1, Kind: "trunk"}, + map[string]meshcatalog.CatalogEntry{"avahi": index.CatalogEntryOf("avahi", stale, catalogue.Manifest{}, graph.Node{None: stale.None2})}) if err != nil { t.Fatal(err) } - held, err := js.KeyValue(ctx, index.ModulesBucket) - if err != nil { - t.Fatal(err) - } - if _, err := held.Put(ctx, "avahi", stale); err != nil { + if err := index.KeepRevision(ctx, &busObjects{js: js, local: index.CatalogRevisionsObjects}, rev, objects); err != nil { t.Fatal(err) } kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.ToolchainsBucket("mesh-build-go"), History: 1}) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 5d60a336..610725e7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -375,9 +375,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - modules, err := process.EssentialBucket(ctx, js, "indexer", index.ModulesBucket, "write a module's fingerprint") - if err != nil { - return err + // The view of the trunk is the newest trunk revision of the catalog graph the objects keep (milestone 47/G, ADR 0656 §6: + // one form of the fact), read once here and kept as a revision after every write; objects not on the bus yet, or none + // kept, start it empty, and the first start's catch-up takes the trunk in once the gate's source is read (catchUp). + revisions := &busObjects{js: js, local: index.CatalogRevisionsObjects} + modules, graph, at, err := index.TrunkView(ctx, revisions) + switch { + case errors.Is(err, process.ErrNotYet): + logger.Printf("the view of the trunk starts empty: %v; the trunk is taken in once they are there and the gate's source is read", err) + modules, graph = index.MemoryStore(), index.MemoryStore() + case err != nil: + return fmt.Errorf("the view of the trunk cannot be read from the newest catalog graph revision: %w", err) + case at == 0: + logger.Printf("the view of the trunk starts empty: %s keeps no trunk revision yet; the trunk is taken in once the gate's source is read", index.CatalogRevisionsObjects) + default: + logger.Printf("the view of the trunk read from trunk revision %d", at) } manifests, err := process.EssentialBucket(ctx, js, "indexer", index.ManifestsBucket, "keep a module's manifest") if err != nil { @@ -413,8 +425,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // What each consumer's go.sum requires of each mesh Go module (DLV-I7): a state this delivery brings, bound when it // appears; until then indexer_requirements answers none, and the builder's floor waits on its answer. requirements := process.AwaitBucket(ctx, js, conn, index.RequirementsBucket, "what consumers require is not kept", logger.Printf) - // The module graph (ADR 0603 §2): a state this delivery brings, bound when it appears; until then no graph is written. - graph := process.AwaitBucket(ctx, js, conn, index.GraphBucket, "no module graph is written", logger.Printf) + // The module graph (ADR 0603 §2) is held in memory, read with the view: each revision's entries carry its nodes. ownBuildState := process.AwaitBucket(ctx, js, conn, index.OwnBuildBucket, "which build this indexer runs is not kept", logger.Printf) // The manifests a take-in wrote and no judge pass judged yet (issue 1966): a state this delivery brings, bound when it // appears; until then a take-in's new manifests are kept for their judging in its memory alone. @@ -427,14 +438,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}} seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) readWhenBound(seatReader, seatDefs, marks) - ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, + ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), 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}, + OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: 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) + Message: messageVia(calls.CallKept), Revisions: revisions, // 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. @@ -728,10 +739,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // said up to; a first view (the tools bundle) answers at that mark. startViews(ctx, js, cred, marks, findings, - map[string]readyOf{"manifests": boundNow(manifests), "modules": boundNow(modules), "candidates": boundNow(candidates), + map[string]readyOf{"manifests": boundNow(manifests), "candidates": boundNow(candidates), index.VerdictsState: awaited(verdicts), index.PredefinedState: awaited(predefined), index.ConfigCandidatesState: awaited(configCandidates), index.PackagesState: awaited(packages), - graphState: awaited(graph), index.HeardState: awaited(heard.b), "taken": awaited(taken), "findings": awaited(findings), + index.HeardState: awaited(heard.b), "taken": awaited(taken), "findings": awaited(findings), index.SeatDefinitionsState: awaited(seatDefs)}, errs, logger.Printf) cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, heard, logger) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 57d8f35d..5d9888dd 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -169,12 +169,15 @@ type Indexer struct { failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes // OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none. OwnBuild func(ctx context.Context) (string, error) - underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) - judgeMu sync.Mutex // one judge pass at a time (verdicts.go) - ownMu sync.Mutex // guards ownFP - ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild) - ownFixed bool // FixOwnBuild asked already: it asks once - ownSince time.Time // when the build was asked and not known, with ownWhy + underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) + judgeMu sync.Mutex // one judge pass at a time (verdicts.go) + ownMu sync.Mutex // guards ownFP + keepMu sync.Mutex // one catalog graph revision kept at a time (keepEntries) + catchingUp atomic.Bool // the first start's catch-up runs: its merges keep no revision until it ends (CatchUp) + catchUpWG sync.WaitGroup // the first start's catch-up in the background (kickCatchUp), for tests to wait on + ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild) + ownFixed bool // FixOwnBuild asked already: it asks once + ownSince time.Time // when the build was asked and not known, with ownWhy ownWhy string // OwnBuildState keeps which build this process knows it runs, or since when and why not (FixOwnBuild); nil keeps none. OwnBuildState Store diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_graph.go b/modules/mesh-delivery-indexer/internal/index/catalog_graph.go index a9fd8e87..d1e6f001 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_graph.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_graph.go @@ -33,6 +33,10 @@ func CatalogEntryOf(module string, cur meshcatalog.ModulesByModuleValue, m catal Commit: cur.Commit, Manifest: cur.Manifest, Verdict: cur.Verdict, Fp: n.FP, None: n.None, Bases: n.Bases, Packages: maps.Clone(cur.Packages), Provides: n.Provides, Requires: n.Requires, Awaits: n.Awaits, Invokes: n.Invokes, Serves: n.Serves, LeftOut: n.LeftOut, Main: MainOf(m)} + if n.FP == "" && n.None == "" { + // A module the recompute computed no node for (its manifest cannot be read) keeps what its value holds. + e.Fp, e.None = cur.Fp2, cur.None2 + } in := meshcatalog.EntryInputs{Toolchains: maps.Clone(cur.Toolchains), Paths: maps.Clone(cur.Paths), Problems: cur.Problems, Warnings: cur.Warnings, Unknown: cur.Unknown, Src: cur.Src, FpSrc: cur.FpSrc} if cur.Inputs2 != nil { @@ -69,6 +73,24 @@ func ModuleValueOf(e meshcatalog.CatalogEntry) meshcatalog.ModulesByModuleValue return v } +// NodeOf is the node of the module graph an entry carries, as CatalogEntryOf took it in: the graph the indexer holds in +// memory is read back from the newest trunk revision at its start (TrunkView), so a revision kept before its first +// recompute loses no edge. +func NodeOf(e meshcatalog.CatalogEntry) graph.Node { + n := graph.Node{V: graph.Version, FP: e.Fp, None: e.None, Repository: e.Repository, Path: e.Path, Bases: e.Bases, + Provides: e.Provides, Requires: e.Requires, Awaits: e.Awaits, Invokes: e.Invokes, Serves: e.Serves, LeftOut: e.LeftOut} + if len(e.Packages) > 0 { + n.Packages = slices.Sorted(maps.Keys(e.Packages)) + } + for language, t := range e.Toolchains { + if n.Toolchains == nil { + n.Toolchains = map[string]graph.Toolchain{} + } + n.Toolchains[language] = graph.Toolchain{Module: t.Module, Version: t.Version, Recipe: t.Recipe} + } + return n +} + // EncodeEntry is an entry's bytes and their sha256 in hex, the name of its object: the same entry in two revisions is // one object (design 47, *An unchanged entry is stored once*). func EncodeEntry(e meshcatalog.CatalogEntry) (string, []byte, error) { diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go index 20a88327..2ccdf931 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "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/catalogue" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -183,14 +184,67 @@ func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graph if err != nil { return err } - entries := make(map[string]meshcatalog.CatalogEntry, len(edges)) - for name, e := range edges { - cur, ok := held[name] - if !ok { - cur = e.cur - } - entries[name] = CatalogEntryOf(name, cur, e.manifest, whole[name]) + // Every module the view holds is an entry, one the recompute computed no edges for (a manifest it cannot read) too. + entries := make(map[string]meshcatalog.CatalogEntry, len(held)) + for name, cur := range held { + entries[name] = CatalogEntryOf(name, cur, edges[name].manifest, whole[name]) } + for name, e := range edges { + if _, ok := held[name]; !ok { + entries[name] = CatalogEntryOf(name, e.cur, e.manifest, whole[name]) + } + } + return ix.keepEntries(ctx, entries) +} + +// KeepView keeps the view of the trunk as a trunk revision after a write to it outside a recompute (the lead's ruling on +// milestone 47/G: the judge pass's verdicts, a build's reported imports, the source stamp, the modules the trunk no longer +// holds, the indexer's own entry a held revision writes): each entry is the module's value in the view with its node of +// the module graph in memory, so what the view holds is never lost at a restart. A view whose entries are the newest +// revision's keeps none. +func (ix *Indexer) KeepView(ctx context.Context) error { + if ix.Revisions == nil || ix.Modules == nil || ix.catchingUp.Load() { + return nil + } + held, err := ix.heldModules(ctx) + if err != nil { + return err + } + nodes := graph.Graph{} + if ix.Graph != nil { + keys, err := ix.Graph.Keys(ctx) + if err != nil && !errors.Is(err, process.ErrNotYet) { + return fmt.Errorf("the module graph: %w", err) + } + for _, k := range keys { + if value, _, found, err := ix.Graph.Get(ctx, k); err == nil && found { + if n, err := graph.Decode(value); err == nil { + nodes[k] = n + } + } + } + } + entries := make(map[string]meshcatalog.CatalogEntry, len(held)) + for name, cur := range held { + var m catalogue.Manifest + if cur.Manifest != "" { + if _, read, err := ix.manifestRaw(ctx, cur.Manifest); err == nil { + m = read + } + } + entries[name] = CatalogEntryOf(name, cur, m, nodes[name]) + } + return ix.keepEntries(ctx, entries) +} + +// keepEntries keeps entries as the trunk revision after the newest, says catalog-revision, and keeps none when no entry +// moved; one keep at a time, so two never take the same number. +func (ix *Indexer) keepEntries(ctx context.Context, entries map[string]meshcatalog.CatalogEntry) error { + if ix.catchingUp.Load() { + return nil // the catch-up of a first start keeps its one revision at its end (CatchUp) + } + ix.keepMu.Lock() + defer ix.keepMu.Unlock() ix.mu.Lock() src := int64(ix.source.rev) ix.mu.Unlock() @@ -209,8 +263,8 @@ func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graph return err } changed := changedEntries(newest.Entries, rev.Entries) - if any && len(changed) == 0 { - return nil + if (any && len(changed) == 0) || (!any && len(entries) == 0) { + return nil // nothing moved; an empty first revision is never kept, so the first start's catch-up still runs } 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) @@ -228,31 +282,46 @@ func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graph // TrunkView is the indexer's view of the trunk read from the newest trunk revision the bucket keeps (the lead's ruling // on milestone 47/G finding 1; ADR 0656 §6, one form of the fact): a store in memory holding each module's value as -// ModuleValueOf reads it back from its entry, under its modules key, and the revision it was read at (0 when the bucket -// keeps none yet: the view starts empty and the first merge taken in fills it). A lost entry is an error naming the -// module (ReadRevision), never a module left out of the view. -func TrunkView(ctx context.Context, o Objects) (Store, int64, error) { - view := newOverlay(nil) +// ModuleValueOf reads it back from its entry, under its modules key, a store in memory of each module's node of the module +// graph (NodeOf), and the revision it was read at (0 when the bucket keeps none yet: both start empty and the first start's +// catch-up fills them). A lost entry is an error naming the module (ReadRevision), never a module left out of the view. +func TrunkView(ctx context.Context, o Objects) (modules, nodes Store, at int64, err error) { + view, held := newOverlay(nil), newOverlay(nil) newest, any, err := newestTrunk(ctx, o) if err != nil || !any { - return view, 0, err + return view, held, 0, err } _, entries, err := ReadRevision(ctx, o, RevisionObject(newest)) if err != nil { - return nil, 0, err + return nil, nil, 0, err } for _, e := range entries { key, err := meshcatalogclient.ModulesKey(e.Module) if err != nil { - return nil, 0, fmt.Errorf("trunk.%d: the entry of %q: %w", newest.Revision, e.Module, err) + return nil, nil, 0, fmt.Errorf("trunk.%d: the entry of %q: %w", newest.Revision, e.Module, err) } data, err := meshcatalogclient.EncodeModule(ModuleValueOf(e)) if err != nil { - return nil, 0, fmt.Errorf("trunk.%d: the entry of %s: %w", newest.Revision, e.Module, err) + return nil, nil, 0, fmt.Errorf("trunk.%d: the entry of %s: %w", newest.Revision, e.Module, err) } if err := view.Create(ctx, key, data); err != nil { - return nil, 0, fmt.Errorf("trunk.%d: the entry of %s: %w", newest.Revision, e.Module, err) + return nil, nil, 0, fmt.Errorf("trunk.%d: the entry of %s: %w", newest.Revision, e.Module, err) + } + gk, err := graph.Key(e.Module) + if err != nil { + continue // a module the graph cannot name has no node, as writeGraph writes none + } + node, err := graph.Encode(NodeOf(e)) + if err != nil { + return nil, nil, 0, fmt.Errorf("trunk.%d: the node of %s: %w", newest.Revision, e.Module, err) + } + if err := held.Create(ctx, gk, node); err != nil { + return nil, nil, 0, fmt.Errorf("trunk.%d: the node of %s: %w", newest.Revision, e.Module, err) } } - return view, newest.Revision, nil + return view, held, newest.Revision, nil } + +// MemoryStore is a store held in memory alone: the module graph's nodes, which each revision's entries carry, so the +// indexer keeps no state of them on the bus (milestone 47/G). +func MemoryStore() Store { return newOverlay(nil) } diff --git a/modules/mesh-delivery-indexer/internal/index/imports.go b/modules/mesh-delivery-indexer/internal/index/imports.go index 50095db6..acd6e152 100644 --- a/modules/mesh-delivery-indexer/internal/index/imports.go +++ b/modules/mesh-delivery-indexer/internal/index/imports.go @@ -27,6 +27,13 @@ import ( // Imports that cannot be resolved (a package the forge cannot list) are kept with no ImportForm: the next recompute that // resolves them writes the import form, one rebuild too many, never a stale module. func (ix *Indexer) ApplyBuildEnd(ctx context.Context, value []byte) error { + if err := ix.applyBuildEnd(ctx, value); err != nil { + return err + } + return ix.KeepView(ctx) // the imports kept with the view (KeepView), so a restart reads them back +} + +func (ix *Indexer) applyBuildEnd(ctx context.Context, value []byte) error { if !builds.IsBuildEntry(value) { return nil } diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 219cbe85..ffe1ffce 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -292,6 +292,13 @@ func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[ // 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 err := ix.reconcile(ctx); err != nil { + return err + } + return ix.KeepView(ctx) // the modules dropped leave the view's next revision too +} + +func (ix *Indexer) reconcile(ctx context.Context) error { if ix.Modules == nil || ix.Trunk == nil { return nil } diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 4756fd1e..c732188f 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -1189,6 +1189,11 @@ func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own } } } + // The indexer's own entry the held revision names is kept as the trunk revision after the newest, every other entry as + // it was (milestone 47/G W4.7: a merged revision naming a newer indexer replaces only the indexer's entry). + if err := ix.KeepView(ctx); err != nil { + return fmt.Errorf("the revision with the indexer's own entry: %w", err) + } t := revisionWait(takein.TakeInPending, ready, names) t.Refused = refused var ended, unmarked []PendingEntry diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 62123afe..8ff4f934 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -2,9 +2,14 @@ package index import ( "context" + "errors" "fmt" "maps" "slices" + "strings" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" @@ -56,6 +61,9 @@ func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) if ix.Modules == nil { return ix.RecomputeCandidates(ctx) } + if err := ix.kickCatchUp(ctx, s.Heads); err != nil { + return err + } // A registered module the forge could not read is read again at each new source revision (issue 1945), in the // background: a forge that does not answer must not hold this revision. One taken in later is stamped when it is written // (applyManifests reads srcOf). @@ -100,6 +108,88 @@ func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) return fmt.Errorf("modules.%s: %w", key, err) } } + // The stamps are kept with the view (KeepView), so a restart reads them back. + if err := ix.KeepView(ctx); err != nil { + return err + } // A trunk that moved makes the previews made onto the old one out of date. return ix.RecomputeCandidates(ctx) } + +// kickCatchUp takes the trunk in at a first start, when the objects keep no trunk revision yet (the lead's ruling A on +// milestone 47/G: the catch-up path): the head the source names for every repository, by the merged message the forge +// bridge gives for it, each applied as its merge, and revision 1 kept from them all at its end. Until then the revision +// the tools answer waits, never an empty catalog. The catch-up runs off the source handler, in its own goroutine, said +// when it ends (lead-47's ruling R1 on the rebase): a bridge that does not answer holds the kept call (ADR 0668) to its +// bound, and must not hold this source revision's stamps and the pulls' candidates with it. A repository whose message is +// not read keeps no revision, said: the next source revision runs the catch-up again (no timer). Objects not on the bus +// yet run none, said, until then. +func (ix *Indexer) kickCatchUp(ctx context.Context, heads map[string]string) error { + if ix.Revisions == nil || ix.Message == nil || len(heads) == 0 { + return nil + } + if _, any, err := newestTrunk(ctx, ix.Revisions); errors.Is(err, process.ErrNotYet) { + ix.logf("catalog graph: the trunk is not taken in yet: its objects %s are not on the bus", CatalogRevisionsObjects) + return nil + } else if err != nil || any { + return err + } + if !ix.catchingUp.CompareAndSwap(false, true) { + return nil + } + heads = maps.Clone(heads) + ix.catchUpWG.Add(1) + go func() { + defer ix.catchUpWG.Done() + unread := ix.takeInHeads(ctx, heads) + ix.catchingUp.Store(false) + if len(unread) > 0 { + ix.logf("catalog graph: the first start's catch-up ended and no first revision is kept: %d repositories were not taken in (%s); taken in again at the next source revision", + len(unread), strings.Join(unread, ", ")) + return + } + if err := ix.KeepView(ctx); err != nil { + ix.logf("catalog graph: the first start's catch-up took in the trunk of %d repositories, but its revision could not be kept: %v; taken in again at the next source revision", len(heads), err) + return + } + ix.logf("catalog graph: the first start's catch-up ended: the trunk of %d repositories taken in", len(heads)) + }() + return nil +} + +// takeInHeads applies the forge bridge's merged message of each repository's head, as its merge, and answers the +// repositories whose message was not read or not applied. +func (ix *Indexer) takeInHeads(ctx context.Context, heads map[string]string) []string { + var unread []string + for _, repository := range slices.Sorted(maps.Keys(heads)) { + head := heads[repository] + if head == "" { + continue + } + f, err := ix.mergedAt(ctx, repository, head) + if err == nil { + _, err = ix.ApplyMessage(ctx, f) + } + if err != nil { + unread = append(unread, repository) + ix.logf("catalog graph: the trunk of %s at %.12s is not taken in: %v", repository, head, err) + } + } + return unread +} + +// mergedAt is the forge bridge's merged message of a repository's head. +func (ix *Indexer) mergedAt(ctx context.Context, repository, head string) (forgemsg.Forge, error) { + data, err := ix.Message(ctx, repository, head, 0) + if err != nil { + return forgemsg.Forge{}, fmt.Errorf("the forge bridge could not answer: %w", err) + } + f, err := forgemsg.DecodeForge(data) + if err != nil { + return f, fmt.Errorf("the forge bridge's answer is not a forge message: %w", err) + } + if f.Kind != forgemsg.ForgeMerged || f.Merge != head || !trunk.SameRepository(f.Repository, repository) { + return f, fmt.Errorf("the forge bridge's answer is not the merge of %s at %s", repository, head) + } + return f, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index e3abfa5e..72366ff3 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -2,6 +2,7 @@ package index import ( "context" + "encoding/json" "errors" "fmt" "maps" @@ -9,6 +10,7 @@ import ( "reflect" "slices" "strings" + "sync" "testing" "time" @@ -354,7 +356,7 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // } // W4.3's second part: the view of the trunk read at start is the newest trunk revision, each module's value as its // entry gives it back; a bucket that keeps none is an empty view at 0. - view, at, err := TrunkView(context.Background(), kept) + view, nodes, at, err := TrunkView(context.Background(), kept) if err != nil || at != 2 { t.Fatalf("the view is read at %d (%v), want 2", at, err) } @@ -365,8 +367,16 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // if want := ModuleValueOf(e); !found || err != nil || !reflect.DeepEqual(got, want) { t.Errorf("the view's %s = %+v (%v), want %+v", name, got, err, want) } + // The node of the module graph the entry carries is read back with the view (NodeOf), so a revision kept before the + // first recompute after a start loses no edge. + gk, _ := graph.Key(name) + rawNode, _, found, _ := nodes.Get(context.Background(), gk) + if back, err := graph.Decode(rawNode); !found || err != nil || !reflect.DeepEqual(NodeOf(e), back) || back.FP != e.Fp || + !reflect.DeepEqual(back.Bases, e.Bases) || !reflect.DeepEqual(CatalogEntryOf(name, ModuleValueOf(e), catalogue.Manifest{}, back).Bases, e.Bases) { + t.Errorf("the view's node of %s = %+v (%v), want %+v", name, back, err, NodeOf(e)) + } } - if empty, at, err := TrunkView(context.Background(), newMemObjects(0)); err != nil || at != 0 || func() bool { k, _ := empty.Keys(context.Background()); return len(k) != 0 }() { + if empty, _, at, err := TrunkView(context.Background(), newMemObjects(0)); err != nil || at != 0 || func() bool { k, _ := empty.Keys(context.Background()); return len(k) != 0 }() { t.Errorf("an empty bucket's view is read at %d (%v), want an empty view at 0", at, err) } _ = kept.Delete(context.Background(), EntryObject(next.Entries[0].Digest)) @@ -445,6 +455,69 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // t.Errorf("%s moved with a build of build-agent: fp2 = %q, want %s", name, m.Fp2, fp) } } + // The lead's ruling A on milestone 47/G: a first start whose objects keep no trunk revision takes in the head the source + // names for every repository, by the forge bridge's merged message, and keeps revision 1 with every module; a message + // the bridge does not give keeps none, and the next source revision takes the trunk in again. + first, _, merged := newFormIndexerUnmerged(t, builderModules(t)...) + first.Revisions = newMemObjects(0) + var firstSaid []meshcatalog.CatalogRevisionEvent + first.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error { + firstSaid = append(firstSaid, e) + return nil + } + // Lead-47's ruling R1: the catch-up runs off the source handler, so a bridge that has not answered yet holds no source + // revision (the first message is answered only once ApplySource returned; held in the handler it is answered as failed). + answers, applied := 0, make(chan struct{}) + first.Message = func(_ context.Context, repository, head string, _ int) ([]byte, error) { + if answers++; answers == 1 { + select { + case <-applied: + return nil, errors.New("the forge bridge does not answer") + case <-time.After(10 * time.Second): + return nil, errors.New("the catch-up held the source handler") + } + } + if repository != catalogueRepo || head != merged.Merge { + return nil, fmt.Errorf("no merge of %s at %s", repository, head) + } + return json.Marshal(merged) + } + heads := meshdelivery.SourceByMainValue{Heads: map[string]string{catalogueRepo: merged.Merge}} + var firstLog []string + var logMu sync.Mutex + first.Logf = func(f string, a ...any) { + logMu.Lock() + firstLog = append(firstLog, fmt.Sprintf(f, a...)) + logMu.Unlock() + } + err = first.ApplySource(context.Background(), 5, heads) + close(applied) + first.catchUpWG.Wait() + if err != nil || len(firstSaid) != 0 || !strings.Contains(strings.Join(firstLog, "\n"), "the forge bridge does not answer") { + t.Fatalf("a catch-up whose message is not read said %+v (%v), log %q, want no revision and the bridge's silence said", firstSaid, err, firstLog) + } + err = first.ApplySource(context.Background(), 6, heads) + first.catchUpWG.Wait() + if err != nil || len(firstSaid) != 1 || firstSaid[0].Revision != 1 || + firstSaid[0].Entries != int64(len(builderModules(t))) || firstSaid[0].Source != 6 { + t.Fatalf("the first start's catch-up said %+v (%v), want trunk revision 1 of source 6 with every module", firstSaid, err) + } + // KeepView: a write to the view outside a recompute (here a verdict carried into one module) is the next revision. + viewKey, _ := meshcatalogclient.ModulesKey("avahi") + _ = write(context.Background(), first.Modules, viewKey, func(value []byte, _ bool) ([]byte, bool, error) { + m, _ := meshcatalogclient.DecodeModule(value) + m.Verdict, m.Problems = meshcatalogclient.VerdictRefused, []string{"a field the judge refuses"} + data, err := meshcatalogclient.EncodeModule(m) + return data, false, err + }) + if err := first.KeepView(context.Background()); err != nil || len(firstSaid) != 2 || !reflect.DeepEqual(firstSaid[1].Changed, []string{"avahi"}) { + t.Errorf("KeepView after a write said %+v (%v), want revision 2 naming avahi", firstSaid, err) + } + if view, _, at, err := TrunkView(context.Background(), first.Revisions); err != nil || at != 2 { + t.Errorf("the view at a restart is read at %d (%v), want 2", at, err) + } else if raw, _, _, _ := view.Get(context.Background(), viewKey); !strings.Contains(string(raw), "a field the judge refuses") { + t.Errorf("the view at a restart lost the verdict: %s", raw) + } // ADR 0614 §2, §3: the graph names what each module invokes and serves, and leaves out, over the whole graph, the // verb edges of two modules calling each other, from either side; a third calling one of them keeps its edge. invoking := func(name, invokes string) mod { diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 24ce8159..22bc5764 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -252,6 +252,10 @@ func (ix *Indexer) judgeOverHeld(ctx context.Context, fresh map[string]bool, ove ix.logf("judge: modules.%s does not carry its verdict: %v", name, err) } } + // The verdicts carried are kept with the view (KeepView), so a restart reads them back. + if err := ix.KeepView(ctx); err != nil { + ix.logf("judge: the verdicts carried are not kept in a catalog graph revision: %v", err) + } if over != nil { return out, nil }