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
This commit is contained in:
@@ -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})
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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) }
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user