novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

B1 (indexer): the indexer's own name is its credential's module, no constant

Browse Source 
index.Module is gone: the indexer and its registry carry the credential's module (findings' observer and ids, its own
entry and graph key, the history subject, the views emitter and reader), the tools take the served-as module.

Builds: design 47 G A-296, §428

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 23:37:54 +01:00
parent 1db4728081
commit 0fffea89b1
43 changed files with 217 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go
+2 -2
View File
@@ -290,7 +290,7 @@ func main() {
tools = append(tools, catalogTools(trunk.State)...)
// The seat mesh-catalog's verb revision, a first view of the catalog half answered in this runtime of its own to exactly
// the parts whose entry declares it (novox/hq ADR 0663).
tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}),
tools = append(tools, index.RevisionTool(revisions, os.Getenv(stdio.ServedAs)), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}),
index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}, Revisions: revisions}),
index.InputsReadTool(keptRead{stdio.State(index.InputsState)}, stateobjects.ForState(index.InputsState).Objects))
tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error {
@@ -325,7 +325,7 @@ func main() {
return err
}, 10*time.Second)...)
tools = append(tools, configReadersTool(stdio.NewWatchedStates(), trunk.State("modules")), busStateTool(trunk.State), requirementsTool(keptSource),
packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(keptSource))
packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(os.Getenv(stdio.ServedAs), keptSource))
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
os.Exit(1)
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go
+6 -5
View File
@@ -19,17 +19,18 @@ import (
// writes into its env value (views.CallersEnv), and its ordinary read tools (design 25 § 12.2). Every one reads only the
// indexer's own states, so it answers as soon as the runtime of its own is connected (ADR 0621 §3).
func viewTools(getenv func(string) string, state func(string) views.Source, now func() time.Time) []stdio.Tool {
module := getenv(stdio.ServedAs) // the module the runtime serves the tools as, never a constant (design 47 G A-296)
marks := state(views.MarksState)
var out []stdio.Tool
for _, t := range index.FirstViewTools {
out = append(out, views.FirstViewTool(index.Module, t.Tool, t.State, t.Words,
out = append(out, views.FirstViewTool(module, t.Tool, t.State, t.Words,
views.ParseCallers(getenv(views.CallersEnv(t.Tool))), state(t.State), marks, now))
}
for _, t := range index.ReadTools {
out = append(out, views.ReadTool(index.Module, t.Tool, t.State, t.Words, state(t.State), marks, now))
out = append(out, views.ReadTool(module, t.Tool, t.State, t.Words, state(t.State), marks, now))
}
// Its own first views, of the assigner's assignments (DLV-F3r): read, and asked again by hand.
return append(out, views.StatusTools(index.Module, "indexer_first_views", "indexer_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, now)...)
return append(out, views.StatusTools(module, "indexer_first_views", "indexer_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, now)...)
}
// keptSource is one of the indexer's states as its runtime reads it.
@@ -171,14 +172,14 @@ func seatDefinitionTool(state func(string) views.Source) stdio.Tool {
}
// outOfStepTool is indexer_out_of_step (design 26; DLV-I19): the generated types out of step with their definitions.
func outOfStepTool(state func(string) views.Source) stdio.Tool {
func outOfStepTool(own string, state func(string) views.Source) stdio.Tool {
return stdio.Tool{Name: index.OutOfStepTool,
Description: "The generated types out of step with their seat definitions (design 26): each trunk contracts module whose " +
"finding generated-types-out-of-step stands (its last trunk build's test step named the files; cleared only by a clean " +
"build of the same module), then each open pull request whose check's verdict named out-of-step files.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
rows, err := index.OutOfStepOf(state(index.SeatDefinitionsState), state(index.FindingsState))
rows, err := index.OutOfStepOf(own, state(index.SeatDefinitionsState), state(index.FindingsState))
if err != nil {
return nil, err
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go
+3 -3
View File
@@ -36,15 +36,15 @@ func TestIndexerAnswersARunsInputOverOneMiBInChunks(t *testing.T) {
sum := sha256.Sum256(body)
digest := hex.EncodeToString(sum[:])
js, _, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.InputsState), History: 1})
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.InputsState), History: 1})
if err != nil {
t.Fatal(err)
}
// The indexer's own seat list (A-186), its state seat-definitions: empty, so the bootstrapped mesh-catalog stands.
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.SeatDefinitionsState), History: 5}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.SeatDefinitionsState), History: 5}); err != nil {
t.Fatal(err)
}
objs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))})
objs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(testIndexer, catalogue.StateObjectsName(index.InputsState))})
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+49 -49
View File
@@ -86,8 +86,8 @@ accounts {
// The beds' indexer's own findings and first views, named from the module its credential names (design 47 G A-296).
var (
findingsBucket = index.Bucket(index.Module, index.FindingsState)
firstViewsBucket = index.Bucket(index.Module, views.StatusState)
findingsBucket = index.Bucket(testIndexer, index.FindingsState)
firstViewsBucket = index.Bucket(testIndexer, views.StatusState)
)
func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
@@ -107,9 +107,9 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
builds.end(t, "novox/mesh-catalog", "c", "modules/c")
}, 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.Bucket(index.Module, index.PendingState): 1, index.Bucket(index.Module, index.VerdictsState): index.VerdictsHistory,
index.Bucket(index.Module, index.TakenState): 1, index.Bucket(index.Module, index.PackagesState): 1,
index.Bucket(index.Module, index.SeatDefinitionsState): 1, index.Bucket(index.Module, index.RequirementsState): 1, findingsBucket: 1, index.Bucket(index.Module, index.UnjudgedState): 1} {
for bucket, history := range map[string]uint8{index.Bucket(testIndexer, index.PendingState): 1, index.Bucket(testIndexer, index.VerdictsState): index.VerdictsHistory,
index.Bucket(testIndexer, index.TakenState): 1, index.Bucket(testIndexer, index.PackagesState): 1,
index.Bucket(testIndexer, index.SeatDefinitionsState): 1, index.Bucket(testIndexer, index.RequirementsState): 1, findingsBucket: 1, index.Bucket(testIndexer, index.UnjudgedState): 1} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: history}); err != nil {
t.Fatal(err)
}
@@ -129,7 +129,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
t.Fatal(err)
}
// The trunk holds the indexer's entry, taken in at an earlier commit with its own manifest; the validation names no Go module (A-268).
seedBuildModules(t, ctx, js, map[string]meshcatalog.CatalogEntry{index.Module: trunkIndexerEntry(t, ctx, js)})
seedBuildModules(t, ctx, js, map[string]meshcatalog.CatalogEntry{testIndexer: trunkIndexerEntry(t, ctx, js)})
validations = standInBuildSeat(t, ctx, js, &refusing)
// The node-engine's composition on n1 (process.Conn.Call's subjects): the build of the indexer that runs there.
answer := func(m *nats.Msg) {
@@ -203,10 +203,10 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
if err != nil || views0.Revision > cur.Revision() || views0.Deleted || len(views0.Value) == 0 || !views0.Created.Equal(bucketCreated(t, ctx, candidates)) {
t.Errorf("views event = %+v, bucket at %d (%v)", views0, cur.Revision(), err)
}
marks, _ := js.KeyValue(ctx, index.Bucket(index.Module, views.MarksState))
f := &views.Follower{Owner: index.Module, Tool: "indexer_read_candidates", State: "candidates",
marks, _ := js.KeyValue(ctx, index.Bucket(testIndexer, views.MarksState))
f := &views.Follower{Owner: testIndexer, Tool: "indexer_read_candidates", State: "candidates",
Ask: func(context.Context, string) (views.Answer, error) {
return views.FirstView(index.Module, "candidates", jsSource{ctx, candidates}, jsSource{ctx, marks}, "", time.Now())
return views.FirstView(testIndexer, "candidates", jsSource{ctx, candidates}, jsSource{ctx, marks}, "", time.Now())
}}
if err := f.Start(ctx); err != nil {
t.Fatal(err)
@@ -228,7 +228,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
if err != nil {
t.Fatal(err)
}
const ix = index.Module
const ix = testIndexer
m1 := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h1,
Merge: "1111111111111111111111111111111111111111", Root: "r0",
Trees: map[string]string{"modules/" + ix: "tree-indexer-2", "modules/a": "tree-a-2"},
@@ -318,17 +318,17 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
if !ok || err != nil || own.Fp2 != named || own.Commit != m1.Merge {
t.Fatalf("modules.%s = %+v (%v), want the named build %s at the merge", ix, own, err, named)
}
if _, ok := get(index.Bucket(index.Module, index.ManifestsState), own.Manifest); !ok {
if _, ok := get(index.Bucket(testIndexer, index.ManifestsState), own.Manifest); !ok {
t.Errorf("manifests.%s is not written", own.Manifest)
}
if _, ok := get(index.Bucket(index.Module, index.VerdictsState), own.Manifest); !ok {
if _, ok := get(index.Bucket(testIndexer, index.VerdictsState), own.Manifest); !ok {
t.Errorf("the verdict on the indexer's manifest %s is not written", own.Manifest)
}
repoKey, _ := trunk.RepositoryKey(m1.Repository)
if _, ok := get(index.ModulesState, "a"); ok {
t.Errorf("modules.a is written while the revision is pending")
}
if _, ok := get(index.Bucket(index.Module, index.TakenState), repoKey); ok {
if _, ok := get(index.Bucket(testIndexer, index.TakenState), repoKey); ok {
t.Errorf("taken.%s is written while the revision is pending", repoKey)
}
if e, err := candidates.Get(ctx, "feat/x"); err != nil {
@@ -351,7 +351,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the group with its first member", groupKey, gathered, ok)
}
// indexer_pending, the tool over the same state, names both waits.
pendingKV, err := js.KeyValue(ctx, index.Bucket(index.Module, index.PendingState))
pendingKV, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.PendingState))
if err != nil {
t.Fatal(err)
}
@@ -388,7 +388,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
if raw, ok := get(index.ModulesState, "a"); !ok || !strings.Contains(string(raw), m1.Merge) {
t.Errorf("modules.a = %s after the take-in at the start, want it at the merge %s", raw, m1.Merge)
}
if _, ok := get(index.Bucket(index.Module, index.TakenState), repoKey); !ok {
if _, ok := get(index.Bucket(testIndexer, index.TakenState), repoKey); !ok {
t.Errorf("taken.%s is not written by the take-in at the start", repoKey)
}
if len(w) != 1 || w[0].Key != groupKey || len(w[0].Keys) != 1 || w[0].Keys[0] != g1Key {
@@ -456,21 +456,21 @@ func trunkIndexerEntry(t *testing.T, ctx context.Context, js jetstream.JetStream
t.Fatal(err)
}
digest := meshcatalogclient.ManifestDigest(raw)
manifests, err := js.KeyValue(ctx, index.Bucket(index.Module, index.ManifestsState))
manifests, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.ManifestsState))
if err != nil {
t.Fatal(err)
}
mkey, _ := meshcatalogclient.ManifestsKey(digest)
mdata, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: index.Module, Manifest: string(raw)})
mdata, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: testIndexer, Manifest: string(raw)})
if _, err := manifests.Put(ctx, mkey, mdata); err != nil {
t.Fatal(err)
}
held := meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-delivery", Commit: "4444444444444444444444444444444444444444", Path: "modules/" + index.Module,
held := meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-delivery", Commit: "4444444444444444444444444444444444444444", Path: "modules/" + testIndexer,
Manifest: digest, None2: sdkstep.ToolchainNoReport}
e := index.CatalogEntryOf(index.Module, held, m, string(raw), graph.Node{None: held.None2})
e := index.CatalogEntryOf(testIndexer, held, m, string(raw), graph.Node{None: held.None2})
for _, a := range m.Build.Artifacts {
e.Artifacts = append(e.Artifacts, meshcatalog.Artifact{Kind: a.Kind, Language: a.Language, Ref: index.ArtifactRefOf(index.Module, a.Name),
Package: &meshcatalog.ArtifactPackage{Ecosystem: index.EcosystemGo, Name: index.Module}})
e.Artifacts = append(e.Artifacts, meshcatalog.Artifact{Kind: a.Kind, Language: a.Language, Ref: index.ArtifactRefOf(testIndexer, a.Name),
Package: &meshcatalog.ArtifactPackage{Ecosystem: index.EcosystemGo, Name: testIndexer}})
}
return e
}
@@ -500,10 +500,10 @@ func standInBuildSeat(t *testing.T, ctx context.Context, js jetstream.JetStream,
t.Fatal(err)
}
// The indexer's state inputs and its objects, where the validation's request is kept for the build seat's holder (§354).
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.InputsState), History: 1}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.InputsState), History: 1}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))}); err != nil {
if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(testIndexer, catalogue.StateObjectsName(index.InputsState))}); err != nil {
t.Fatal(err)
}
s := &standIn{}
@@ -737,12 +737,12 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
t.Fatal(err)
}
}
for _, bucket := range []string{index.Bucket(index.Module, index.PinsState), index.Module + "_" + process.ConsumersState} {
for _, bucket := range []string{index.Bucket(testIndexer, index.PinsState), testIndexer + "_" + process.ConsumersState} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil {
t.Fatal(err)
}
}
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.NamedState), History: 1}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.NamedState), History: 1}); err != nil {
t.Fatal(err)
}
// The build modules' streams for toolchain.reported (declared by them, heard by the indexer) and its durable consumers.
@@ -754,7 +754,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
t.Fatal(err)
}
}
candidates, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.CandidatesState), History: index.CandidatesHistory})
candidates, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.CandidatesState), History: index.CandidatesHistory})
if err != nil {
t.Fatal(err)
}
@@ -764,7 +764,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
}
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.Bucket(index.Module, index.ManifestsState), History: index.ManifestsHistory}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.ManifestsState), History: index.ManifestsHistory}); err != nil {
t.Fatal(err)
}
// The builder's module log, as the runtime raises it (ADR 0297), filled by put before the indexer starts.
@@ -774,7 +774,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
}
// The indexer's marks (ADR 0615 §5), and a waiter on manifest-accepted, which the views family never wakes (its events
// are ordinary events on the shared stream, ADR 0635 §5).
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, views.MarksState), History: 1}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, views.MarksState), History: 1}); err != nil {
t.Fatal(err)
}
acceptedHeard.Store(0)
@@ -797,8 +797,8 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
}
// The indexer's own stream of its findings' events (design 47 G §383), which its finding writer says into (A-30).
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: findingsStream, Subjects: []string{
"mesh.stream." + index.Module + ".event.finding-written.*", "mesh.stream." + index.Module + ".event.finding-cleared.*",
"mesh.stream." + index.Module + ".event.findings-stated.*"}}); err != nil {
"mesh.stream." + testIndexer + ".event.finding-written.*", "mesh.stream." + testIndexer + ".event.finding-cleared.*",
"mesh.stream." + testIndexer + ".event.findings-stated.*"}}); err != nil {
t.Fatal(err)
}
put(ctx, buildLog{ctx: ctx, js: js})
@@ -847,14 +847,14 @@ const buildSeatVerbs = `{"name":"current","input":{"type":"object"},"answer":{"t
// revision (revision 1) holding both holders beside others, the entries a test's own fixture holds.
func seedBuildModules(t *testing.T, ctx context.Context, js jetstream.JetStream, others map[string]meshcatalog.CatalogEntry) {
t.Helper()
defs, err := js.KeyValue(ctx, index.Bucket(index.Module, index.SeatDefinitionsState))
defs, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.SeatDefinitionsState))
if errors.Is(err, jetstream.ErrBucketNotFound) {
defs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.SeatDefinitionsState), History: 1})
defs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.SeatDefinitionsState), History: 1})
}
if err != nil {
t.Fatal(err)
}
manifests, err := js.KeyValue(ctx, index.Bucket(index.Module, index.ManifestsState))
manifests, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.ManifestsState))
if err != nil {
t.Fatal(err)
}
@@ -888,7 +888,7 @@ func seedBuildModules(t *testing.T, ctx context.Context, js jetstream.JetStream,
t.Fatal(err)
}
// A holder the mesh took in has its accepted verdict: the indexer judges and says nothing of it at its start.
if verdicts, err := js.KeyValue(ctx, index.Bucket(index.Module, index.VerdictsState)); err == nil {
if verdicts, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.VerdictsState)); err == nil {
vkey, err := meshcatalogclient.VerdictsKey(digest)
if err != nil {
t.Fatal(err)
@@ -923,8 +923,8 @@ func seedBuildModules(t *testing.T, ctx context.Context, js jetstream.JetStream,
// serves.
func seedClaimant(t *testing.T, ctx context.Context, js jetstream.JetStream, manifests jetstream.KeyValue, entries map[string]meshcatalog.CatalogEntry) {
t.Helper()
if _, err := js.KeyValue(ctx, index.Bucket(index.Module, index.SeatDefinitionsState)); errors.Is(err, jetstream.ErrBucketNotFound) {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.SeatDefinitionsState), History: 1}); err != nil {
if _, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.SeatDefinitionsState)); errors.Is(err, jetstream.ErrBucketNotFound) {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.SeatDefinitionsState), History: 1}); err != nil {
t.Fatal(err)
}
}
@@ -954,7 +954,7 @@ func seedClaimantAlone(t *testing.T, ctx context.Context, js jetstream.JetStream
if list, err := index.ListRevisions(ctx, objects, false); err != nil || len(list.Revisions) > 0 {
return
}
manifests, err := js.KeyValue(ctx, index.Bucket(index.Module, index.ManifestsState))
manifests, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.ManifestsState))
if err != nil {
t.Fatal(err)
}
@@ -973,7 +973,7 @@ func seedClaimantAlone(t *testing.T, ctx context.Context, js jetstream.JetStream
// §699): the indexer consumes it in the seat's form, and its own manifest is judged against the seat list it holds.
func seedAssignerSeat(t *testing.T, ctx context.Context, js jetstream.JetStream) {
t.Helper()
defs, err := js.KeyValue(ctx, index.Bucket(index.Module, index.SeatDefinitionsState))
defs, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.SeatDefinitionsState))
if err != nil {
t.Fatal(err)
}
@@ -1024,7 +1024,7 @@ func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the m
builds.end(t, "novox/mesh-controller", "build-agent", "modules/build-agent")
}, func(ctx context.Context, js jetstream.JetStream) {
var err error
if asks, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.TakeInsState), History: 1}); err != nil {
if asks, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.TakeInsState), History: 1}); err != nil {
t.Fatal(err)
}
})
@@ -1134,7 +1134,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
}, func(ctx context.Context, js jetstream.JetStream) {
// The indexer's state unjudged, as the controller makes it from its manifest; the verdicts state is made only once
// the merge is taken in (issue 1966 note 154: a take-in at the start, before the verdicts state's bind).
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.UnjudgedState), History: 1}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.UnjudgedState), History: 1}); err != nil {
t.Fatal(err)
}
})
@@ -1166,7 +1166,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
m.Inputs2.Locks["tools"] != goSum || m.Path != "modules/avahi" { // 1606/AC2
t.Errorf("module = %+v within 20 s, want its path modules/avahi, its manifest digest %s, its inputs and none2 'toolchain not yet reported'", m, digest)
}
kv, err := js.KeyValue(ctx, index.Bucket(index.Module, index.ManifestsState))
kv, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.ManifestsState))
if err != nil {
t.Fatal(err)
}
@@ -1181,7 +1181,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
// 1966 (ADR 0668 §4, note 154): the take-in wrote its four manifests with no verdicts state on the bus, so it keeps
// them in its state unjudged; when the verdicts state binds, each is judged as new (bad refused in full, never as a
// warning), every accepted one is said as manifest-accepted, and the state is left empty.
unjudged, err := js.KeyValue(ctx, index.Bucket(index.Module, index.UnjudgedState))
unjudged, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.UnjudgedState))
if err != nil {
t.Fatal(err)
}
@@ -1191,7 +1191,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
t.Errorf("the take-in's new manifest %s is not kept unjudged: the state holds %v", meshcatalogclient.ManifestDigest([]byte(raw)), kept)
}
}
verdicts, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.VerdictsState), History: index.VerdictsHistory})
verdicts, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.VerdictsState), History: index.VerdictsHistory})
if err != nil {
t.Fatal(err)
}
@@ -1346,10 +1346,10 @@ func TestIndexerBuildsARunsCandidateAndDropsItWhenEveryMembersPipelineEnded(t *t
}
}
var err error
if runs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.RunsState), History: 1}); err != nil {
if runs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.RunsState), History: 1}); err != nil {
t.Fatal(err)
}
regs, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.RegistrationsState), History: 1})
regs, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.RegistrationsState), History: 1})
if err != nil {
t.Fatal(err)
}
@@ -1436,7 +1436,7 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC
if err != nil {
t.Fatal(err)
}
manifests, err := js.KeyValue(ctx, index.Bucket(index.Module, index.ManifestsState))
manifests, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.ManifestsState))
if err != nil {
t.Fatal(err)
}
@@ -1681,7 +1681,7 @@ func TestIndexerKeepsWhatAWantAndARunningBuildNameAndLetsAGoneOneLeave(t *testin
}
}
})
named, err := js.KeyValue(ctx, index.Bucket(index.Module, index.NamedState))
named, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.NamedState))
if err != nil {
t.Fatal(err)
}
@@ -1722,11 +1722,11 @@ func TestIndexerAppliesThePinsEventsThroughItsDurableConsumer(t *testing.T) {
}
say(meshdelivery.SubjectPinned, meshdelivery.Pins{Record: "r1", Milestone: "47/G", Catalog: []meshdelivery.PinsCatalogItem{{Branch: "main", Revision: 9}}})
say(meshdelivery.SubjectPinsStated, meshdelivery.PinsStatedEvent{At: time.Now().UTC().Add(time.Minute)})
pins, err := js.KeyValue(ctx, index.Bucket(index.Module, index.PinsState))
pins, err := js.KeyValue(ctx, index.Bucket(testIndexer, index.PinsState))
if err != nil {
t.Fatal(err)
}
consumers, err := js.KeyValue(ctx, index.Module+"_"+process.ConsumersState)
consumers, err := js.KeyValue(ctx, testIndexer+"_"+process.ConsumersState)
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+5 -5
View File
@@ -457,12 +457,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return fmt.Errorf("answering the verb input: %w", err)
}
defer stopInput()
ix := &index.Indexer{NamedSets: named, Findings: writer, 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,
ix := &index.Indexer{Module: cred.Module, NamedSets: named, Findings: writer, 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.Bucket(cred.Module, index.PackagesState)}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
OwnBuild: ownBuildVia(calls.Call, cred.Node), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph,
OwnBuild: ownBuildVia(calls.Call, cred.Module, cred.Node), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph,
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.Bucket(cred.Module, index.EntryIndexState), "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
// Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go).
@@ -558,7 +558,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered
// before it starts.
hub := process.NewEventHub(logger.Printf)
reader := &views.ProcessReader{Module: index.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub,
reader := &views.ProcessReader{Module: cred.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub,
Findings: func() views.WireWriter { return writer }}
// The holder function (A-302): the seat-holders view of the claimant of mesh-module-assigner in the indexer's own catalog,
// followed again at each refresh of the build modules (a seat list change, a catalog revision).
@@ -638,7 +638,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// Through the seat git (milestone 47/G W4.15, design 47 *A part that cannot read the forge keeps what it has*): a seat that
// does not answer is waited for at most index.GitWaitBound before the judging below runs on the kept definitions, said in
// indexer_first_views and as view-waits past five minutes, and read again on the next forge message heard.
defs := gitWaits(ctx, ix, writer, cred.Node, process.AwaitBucket(ctx, js, conn, index.Bucket(cred.Module, views.StatusState), "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf)
defs := gitWaits(ctx, cred.Module, ix, writer, cred.Node, process.AwaitBucket(ctx, js, conn, index.Bucket(cred.Module, views.StatusState), "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf)
seatDefs.Ready(func(jetstream.KeyValue) { defs.Start(ctx) })
// Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578
// §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage).
@@ -674,7 +674,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// with the build module, as with the gate's source.
// Each consumer below that is not there yet, or cannot be heard, is the finding family-not-heard naming its stream until
// it binds at a later start (heard.Writer; ADR 0615 §5), so the indexer is not healthy while what it waits on is lost.
notHeard := notHeardWriter(writer, logger.Printf)
notHeard := notHeardWriter(cred.Module, writer, logger.Printf)
// A consumer that is there and cannot be heard stops the indexer, as before; it is said as the finding first.
var stopped error
hear := func(stream, consequence string, each func(jetstream.Msg)) {
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
+4 -1
View File
@@ -78,7 +78,10 @@ func (m *memStore) Delete(_ context.Context, k string, _ uint64) error { delete(
const head = "0123456789abcdef0123456789abcdef01234567"
// testAccount is the bus account the tests run the indexer as.
var testAccount = process.Account{Node: "n1", Module: "mesh-delivery-indexer"}
// testIndexer is the module the bed's credential names; the indexer names its own buckets, findings and events from it (A-296).
const testIndexer = "mesh-delivery-indexer"
var testAccount = process.Account{Node: "n1", Module: testIndexer}
func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
// Review round 2: the indexer's views emitter writes its finding view-unsaid (Emitter.Findings set).
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go
+6 -6
View File
@@ -101,13 +101,13 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
t.Errorf("a failed call: %v, want an error that is not ErrNotInTree", err)
}
// The build the indexer runs is the engine's word (Runs.FP of its composition verb), asked of its own machine.
var asked string
var asked, askedFor string
own := func(key string, body any) (json.RawMessage, error) {
asked = key
asked, askedFor = key, body.(map[string]any)["module"].(string)
return json.RawMessage(`{"runs":{"fp":"src2:abc"}}`), nil
}
if fp, err := ownBuildVia(own, "n1")(context.Background()); err != nil || fp != "src2:abc" || asked != "seat:node-engine.composition@n1" {
t.Errorf("own build = %q %v asked %q", fp, err, asked)
if fp, err := ownBuildVia(own, "acme-indexer", "n1")(context.Background()); err != nil || fp != "src2:abc" || asked != "seat:node-engine.composition@n1" || askedFor != "acme-indexer" {
t.Errorf("own build = %q %v asked %q for %q", fp, err, asked, askedFor)
}
// Issue 1921: the start during a replacement. What runs still names the old build, the want (composed by the node itself)
// the successor, which this process is: it fixes the successor. A want that is not self, or is removed, leaves what runs.
@@ -121,12 +121,12 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
{"no build runs yet, want self", `{"want":{"fp":"src2:new","self":true}}`, "src2:new"},
} {
ask := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil }
if fp, err := ownBuildVia(ask, "n1")(context.Background()); err != nil || fp != row.want {
if fp, err := ownBuildVia(ask, "acme-indexer", "n1")(context.Background()); err != nil || fp != row.want {
t.Errorf("%s: own build = %q %v, want %q", row.name, fp, err, row.want)
}
}
none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{}`), nil }
if _, err := ownBuildVia(none, "n1")(context.Background()); err == nil {
if _, err := ownBuildVia(none, "acme-indexer", "n1")(context.Background()); err == nil {
t.Error("an engine that says nothing runs gave a build")
}
// Design 47 G §64, A-160, A-174: the runner's run-started and pipeline-ended reach the indexer; one that is no such body,
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/ownbuild.go
+2 -4
View File
@@ -5,8 +5,6 @@ import (
"encoding/json"
"errors"
"fmt"
"mesh-delivery-indexer/internal/index"
)
// compositionVerb is the node-engine seat's read verb that says what runs of a module on a machine (its Runs.FP), asked of
@@ -15,9 +13,9 @@ import (
const compositionVerb = "seat:node-engine.composition"
// ownBuildVia asks the engine of node which fingerprint of the indexer runs there.
func ownBuildVia(call toolCaller, node string) func(ctx context.Context) (string, error) {
func ownBuildVia(call toolCaller, module, node string) func(ctx context.Context) (string, error) {
return func(context.Context) (string, error) {
raw, err := call(compositionVerb+"@"+node, map[string]any{"module": index.Module})
raw, err := call(compositionVerb+"@"+node, map[string]any{"module": module})
if err != nil {
return "", fmt.Errorf("%s: %w", compositionVerb, err)
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go
+5 -5
View File
@@ -65,10 +65,10 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
var packages, seatDefs, findings, takeIns jetstream.KeyValue
js, ctx, _, modules := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) {
var err error
if packages, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.PackagesState), History: index.PackagesHistory}); err != nil {
if packages, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.PackagesState), History: index.PackagesHistory}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.TakenState), History: 1}); err != nil {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.TakenState), History: 1}); err != nil {
t.Fatal(err)
}
// Both languages' toolchains, so mesh-sdk (a TypeScript package) gets a fingerprint that moves with its source.
@@ -85,7 +85,7 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
}
serveForge(t, js.Conn(), listing, goSumContent)
// The seat definitions' state (DLV-I17), as the grants part makes it from the manifest.
if seatDefs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.SeatDefinitionsState), History: 5}); err != nil {
if seatDefs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.SeatDefinitionsState), History: 5}); err != nil {
t.Fatal(err)
}
// DLV-I19, I20: the indexer's findings and takeins states, and the checker's stream asks with the indexer's durable
@@ -93,7 +93,7 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
if findings, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: findingsBucket, History: 1}); err != nil {
t.Fatal(err)
}
if takeIns, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(index.Module, index.TakeInsState), History: 1}); err != nil {
if takeIns, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.Bucket(testIndexer, index.TakeInsState), History: 1}); err != nil {
t.Fatal(err)
}
seedBuildModules(t, ctx, js, nil)
@@ -322,7 +322,7 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
t.Helper()
var rows []index.OutOfStepRow
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if rows, err = index.OutOfStepOf(jsSource{ctx, seatDefs}, jsSource{ctx, findings}); err == nil && len(rows) == want {
if rows, err = index.OutOfStepOf(testIndexer, jsSource{ctx, seatDefs}, jsSource{ctx, findings}); err == nil && len(rows) == want {
return rows
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/registry.go
+6 -6
View File
@@ -43,21 +43,21 @@ func publishAs(conn *nats.Conn, cred process.Account, subject, id string, body a
return conn.PublishMsg(msg)
}
// historySubject is where one module's versions land on the stream registration-history.
func historySubject(module string) string {
return process.ModuleStreamSubject(index.Module, index.RegistrationHistoryEvent+"."+module)
// historySubject is where one module (of the indexer's own module)'s versions land on the stream registration-history.
func historySubject(own, module string) string {
return process.ModuleStreamSubject(own, index.RegistrationHistoryEvent+"."+module)
}
// newRegistry is the Registry over the indexer's own states, the stream registration-history, the seat git, and the seat
// operator-channel's accepts. where is the read of the router's own record of an ask (routerRecordOf, §700).
func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account, call toolCaller, registrations, proposals index.Store, where func(id string) (string, *oc.Warrant, error), ix *index.Indexer, logf func(string, ...any)) *index.Registry {
return &index.Registry{Registrations: registrations, Proposals: proposals,
return &index.Registry{Module: cred.Module, Registrations: registrations, Proposals: proposals,
History: func(ctx context.Context, r meshcatalog.Registration) error {
data, err := json.Marshal(r)
if err != nil {
return err
}
_, err = js.Publish(ctx, historySubject(r.Module), data, jetstream.WithMsgID(fmt.Sprintf("%s.%d", r.Module, r.Version)))
_, err = js.Publish(ctx, historySubject(cred.Module, r.Module), data, jetstream.WithMsgID(fmt.Sprintf("%s.%d", r.Module, r.Version)))
if err == nil {
// Past 80 % full the stream's fill is a finding (design 47 G §704).
if st, serr := js.Stream(ctx, catalogue.StreamBusName(index.RegistrationHistoryStream)); serr == nil {
@@ -69,7 +69,7 @@ func newRegistry(js jetstream.JetStream, conn *nats.Conn, cred process.Account,
return err
},
HistoryOf: func(ctx context.Context, module string) ([]meshcatalog.Registration, error) {
cons, err := js.OrderedConsumer(ctx, catalogue.StreamBusName(index.RegistrationHistoryStream), jetstream.OrderedConsumerConfig{FilterSubjects: []string{historySubject(module)}})
cons, err := js.OrderedConsumer(ctx, catalogue.StreamBusName(index.RegistrationHistoryStream), jetstream.OrderedConsumerConfig{FilterSubjects: []string{historySubject(cred.Module, module)}})
if err != nil {
return nil, err
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go
+8 -8
View File
@@ -51,7 +51,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun
// viewsEmitter is the indexer's emitter of its views family: a change not said past views.Bound is its finding
// view-unsaid, written through its finding writer (review round 2).
func viewsEmitter(js jetstream.JetStream, cred process.Account, mk jetstream.KeyValue, writer views.WireWriter, logf func(string, ...any)) views.Emitter {
return views.Emitter{Owner: index.Module, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js),
return views.Emitter{Owner: cred.Module, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js),
Marks: mk, Findings: writer, Logf: logf}
}
@@ -71,8 +71,8 @@ func sortedStates(states map[string]readyOf) []string {
}
// notHeardWriter writes the indexer's finding of a family it does not hear through its finding writer (design 47 G §689).
func notHeardWriter(w index.FindingWriter, logf func(string, ...any)) heard.Writer {
return heard.Writer{Observer: index.Module, Logf: logf,
func notHeardWriter(module string, w index.FindingWriter, logf func(string, ...any)) heard.Writer {
return heard.Writer{Observer: module, Logf: logf,
Put: func(_ string, value []byte) error {
var f mcseat.Finding
if err := json.Unmarshal(value, &f); err != nil {
@@ -162,17 +162,17 @@ func followAssignments(ctx context.Context, reader *views.ProcessReader, holders
// gitWaits is the indexer's read of the trunk definitions at its start through the seat git (index.DefinitionsAtStart): its
// wait said as a first view of the seat's (Owner git, State seat-definitions) in the indexer's first views, and its finding
// view-waits written past views.WaitsBound and cleared once the seat answers, by hand through the finding writer (design 47 G A-28).
func gitWaits(ctx context.Context, ix *index.Indexer, w index.FindingWriter, node string, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart {
func gitWaits(ctx context.Context, module string, ix *index.Indexer, w index.FindingWriter, node string, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart {
finding := func(s views.Status, raised bool) {
id := keys.EscapeName(index.Module + "." + s.Owner + "." + s.State)
id := keys.EscapeName(module + "." + s.Owner + "." + s.State)
var err error
if raised {
at := time.Now().UTC()
summary := fmt.Sprintf("%s waits for the view %s.%s at revision %s since %s and acts on nothing new until it holds it; it holds %s",
index.Module, s.Owner, s.State, s.WaitsFor, s.WaitingSince.UTC().Format(time.RFC3339), cmp.Or(s.View, "none"))
module, s.Owner, s.State, s.WaitsFor, s.WaitingSince.UTC().Format(time.RFC3339), cmp.Or(s.View, "none"))
err = w.Write(ctx, mcseat.Finding{Scope: views.FindingScope, ID: id, Kind: views.ViewWaitsKind, Severity: mcseat.FindingSeverityWarning,
Summary: summary, Headline: index.Module + " waits for a view of " + s.Owner, Source: "views", Observer: index.Module, At: at,
Subject: mcseat.Subject{Scope: views.FindingScope, ID: index.Module, Machine: node}})
Summary: summary, Headline: module + " waits for a view of " + s.Owner, Source: "views", Observer: module, At: at,
Subject: mcseat.Subject{Scope: views.FindingScope, ID: module, Machine: node}})
} else {
err = clearHeld(ctx, w, views.FindingScope, id, views.ViewWaitsKind, "the part no longer observes it")
}
modules/mesh-delivery-indexer/internal/index/accepted.go
+4 -4
View File
@@ -61,7 +61,7 @@ func (ix *Indexer) sayAccepted(ctx context.Context, a meshcatalog.ManifestAccept
}
// unsaidID is the id of module's unsaid finding.
func unsaidID(module string) string { return Module + ".accepted." + module }
func unsaidID(own, module string) string { return own + ".accepted." + module }
// unsaid raises module's finding for err, or clears it with none.
func (ix *Indexer) unsaid(ctx context.Context, module string, err error) {
@@ -70,9 +70,9 @@ func (ix *Indexer) unsaid(ctx context.Context, module string, err error) {
}
var werr error
if err == nil {
werr = ix.clearFinding(ctx, mcc.ScopeModule, unsaidID(module), UnsaidKind, "manifest-accepted of "+module+" was said.")
werr = ix.clearFinding(ctx, mcc.ScopeModule, unsaidID(ix.Module, module), UnsaidKind, "manifest-accepted of "+module+" was said.")
} else {
werr = ix.Findings.Write(ctx, ix.observed(mcc.ScopeModule, unsaidID(module), UnsaidKind,
werr = ix.Findings.Write(ctx, ix.observed(mcc.ScopeModule, unsaidID(ix.Module, module), UnsaidKind,
fmt.Sprintf("%s of %s was not published: %v", meshcatalog.EventManifestAccepted, module, err),
"The indexer could not tell the bus module that "+module+"'s manifest was accepted",
"The bus module creates and removes a module's bus state from this event. The indexer says it again at its "+
@@ -152,7 +152,7 @@ func (ix *Indexer) SayUnsaid(ctx context.Context) {
if f.Kind != UnsaidKind {
continue
}
module := strings.TrimPrefix(f.Subject.ID, Module+".accepted.")
module := strings.TrimPrefix(f.Subject.ID, ix.Module+".accepted.")
if _, err := ix.sayFor(ctx, module, true, false); err != nil {
ix.logf("judge: %v; its finding stands", err)
}
modules/mesh-delivery-indexer/internal/index/branch_builds_test.go
+1 -1
View File
@@ -16,7 +16,7 @@ import (
// keys when it is no longer watched.
func TestABuildOfAnotherWatchedBranchKeepsItsEdgesAndIsOutdatedAgainstThatBranch(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), Heard: newMem(), BuildEdges: newMem(), Revisions: newMemObjects(1 << 20), Registry: &Registry{Registrations: MemoryStore()}}
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), BuildEdges: newMem(), Revisions: newMemObjects(1 << 20), Registry: &Registry{Registrations: MemoryStore()}}
for _, m := range []string{"lib", "app"} {
raw, _ := json.Marshal(meshcatalog.Registration{Module: m, Branches: []string{"main", "beta"}})
if err := ix.Registry.Registrations.Create(ctx, RegistrationKey(m), raw); err != nil {
modules/mesh-delivery-indexer/internal/index/candidates.go
+3 -1
View File
@@ -55,7 +55,9 @@ const writeTries = 8
// Indexer writes candidates.<group> from the forge bridge's messages.
type Indexer struct {
Store Store // candidates
// Module is the indexer's own module, the credential's: the observer of its findings and the key of its own entry (design 47 G A-296).
Module string
Store Store // candidates
// Modules is the modules bucket; nil writes no fingerprints.
Modules Store
// Manifests is the manifests bucket; nil writes no manifests (issue 1449).
modules/mesh-delivery-indexer/internal/index/catalog_branches_test.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
// branch it is asked for; the event names the branch, escaped.
func TestACatalogRevisionIsKeptPerWatchedBranch(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Revisions: newMemObjects(1 << 20), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
var said []meshcatalog.CatalogRevisionEvent
ix.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error { said = append(said, e); return nil }
reader := meshcatalog.CatalogEntry{Module: "reader", Repository: "novox/r", Commit: "c1", Invokes: []string{RevisionInvoke}, Manifest: map[string]any{}}
modules/mesh-delivery-indexer/internal/index/catalog_edges_test.go
+2 -2
View File
@@ -50,7 +50,7 @@ func TestAnEntryCarriesItsEdgesOfKindsOneToFourAndSix(t *testing.T) {
// Design 47 G A-208: a revision kept holds each entry with its edges, so the builder and the assigner read them from the entry.
func TestAKeptRevisionHoldsTheEdgesOfItsEntries(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Revisions: newMemObjects(1 << 20), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
err := ix.KeepBranchEntries(ctx, "main", 1, map[string]meshcatalog.CatalogEntry{
"b": {Module: "b", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "b:p"}}, Serves: []string{"seat:s.v"}},
"a": {Module: "a", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "a:p"}}, Invokes: []string{"seat:s.v"}},
@@ -202,7 +202,7 @@ func TestAKeptRevisionHoldsTheActsAndTheKindFiveEdges(t *testing.T) {
ctx := context.Background()
defs := newMem()
_ = defs.Create(ctx, "mesh-grants.trunk", []byte(`{"seat":"mesh-grants","source":"trunk","digest":"d","definition":{"seat":"mesh-grants","acts":[{"step":"grant","on":"every"}]}}`))
ix := &Indexer{Revisions: newMemObjects(1 << 20), SeatDefinitions: defs, Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), SeatDefinitions: defs, Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
err := ix.KeepBranchEntries(ctx, "main", 1, map[string]meshcatalog.CatalogEntry{
"grants": {Module: "grants", Manifest: map[string]any{"module": "grants", "version": "1.0", "claims": []any{map[string]any{"name": "mesh-grants"}}}, Artifacts: []meshcatalog.Artifact{{Ref: "grants:p"}}},
"app": {Module: "app", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "app:p"}}},
modules/mesh-delivery-indexer/internal/index/catalog_graph_test.go
+2 -2
View File
@@ -35,7 +35,7 @@ func TestAnEntryCarriesItsManifestWholeAndTheDigestOfItsBytes(t *testing.T) {
// Design 47 G §676: at take-in the entry's artifacts carry the package name read from the revision's own files at the
// artifact's path: the module of a go.mod, the name of a package.json.
func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) {
ix := &Indexer{ReadFile: func(_ context.Context, repository, sha, path string) ([]byte, error) {
ix := &Indexer{Module: testSelf, ReadFile: func(_ context.Context, repository, sha, path string) ([]byte, error) {
files := map[string]string{
"mods/m/go/go.mod": "module git.example/m/go\n\ngo 1.26\n",
"mods/m/js/package.json": `{"name":"@example/m"}`,
@@ -72,7 +72,7 @@ func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) {
// Design 47 G §694: an image artifact's package is {image, <module>/<artifact>}, read from no file, so two modules
// with an image artifact of one name carry two package names.
func TestAnImageArtifactsPackageNameIsItsModulesAndItsOwn(t *testing.T) {
ix := &Indexer{}
ix := &Indexer{Module: testSelf}
names := map[string]bool{}
for _, module := range []string{"a", "b"} {
m := catalogue.Manifest{Module: module, Build: &catalogue.Build{Artifacts: []catalogue.Artifact{{Name: "server", Kind: catalogue.ArtifactImage}}}}
modules/mesh-delivery-indexer/internal/index/control_records_test.go
+7 -7
View File
@@ -96,7 +96,7 @@ func TestTheControlRecords(t *testing.T) {
// highest last, a pseudo-version below the release it precedes.
t.Run("indexer_requirements", func(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Requirements: newMem()}
ix := &Indexer{Module: testSelf, Requirements: newMem()}
sdk := "git.novox.be/novox/mesh-sdk/go"
sum := func(lines ...string) []byte { return []byte(strings.Join(lines, "\n")) }
ix.keepRequirements(ctx, "novox/a", "modules/x", sum(
@@ -167,7 +167,7 @@ func TestTheControlRecords(t *testing.T) {
}
defs, findings := newMem(), newFindings()
var said []SeatDefinitionChanged
ix := &Indexer{SeatDefinitions: defs, Findings: findings, Logf: t.Logf,
ix := &Indexer{Module: testSelf, SeatDefinitions: defs, Findings: findings, Logf: t.Logf,
// The forge's listing of a contracts module's `seats/` directory: every blob under it at the commit (ADR 0635 §6).
Files: func(_ context.Context, _, sha string, dirs []string) (map[string]map[string]string, error) {
out := map[string]map[string]string{}
@@ -204,7 +204,7 @@ func TestTheControlRecords(t *testing.T) {
}
return e, ok
}
unparsed := mcc.Key(mcc.ScopeModule, findingID("seat-definitions", "mesh-contracts"), SeatDefinitionUnparsedKind)
unparsed := mcc.Key(mcc.ScopeModule, findingID(testSelf, "seat-definitions", "mesh-contracts"), SeatDefinitionUnparsedKind)
merged(m1)
merged(m1) // the same digests again say nothing
g1, ok := entry("git.trunk")
@@ -252,7 +252,7 @@ func TestTheControlRecords(t *testing.T) {
t.Fatal(err)
}
}
twice := mcc.Key(mcc.ScopeModule, findingID("seat-defined-twice", "other-contracts"), SeatDefinedTwiceKind)
twice := mcc.Key(mcc.ScopeModule, findingID(testSelf, "seat-defined-twice", "other-contracts"), SeatDefinedTwiceKind)
before := len(said)
mergedOther("o1")
if g, _ := entry("git.trunk"); g.Repository != repo || g.Commit != m3 || len(said) != before {
@@ -302,7 +302,7 @@ func TestTheControlRecords(t *testing.T) {
for k, v := range findings.vals {
fsrc[k] = &views.Entry{Value: v}
}
rows, err := OutOfStepOf(src(), fsrc)
rows, err := OutOfStepOf(testSelf, src(), fsrc)
if err != nil || len(rows) != 2 || rows[0].Module != "mesh-contracts" || !strings.Contains(rows[0].Summary, "ts/x.ts") ||
rows[1].Pull != repo+"#9" || !reflect.DeepEqual(rows[1].Files, []string{"go/a.go", "go/b.go"}) {
t.Errorf("out of step %+v, %v; want the trunk's module and the pull request's sorted files", rows, err)
@@ -402,7 +402,7 @@ func TestTheControlRecords(t *testing.T) {
t.Fatal(err)
}
}
outside := mcc.Key(mcc.ScopeModule, findingID("seat-outside-its-domain", "mesh-contracts"), SeatOutsideItsDomainKind)
outside := mcc.Key(mcc.ScopeModule, findingID(testSelf, "seat-outside-its-domain", "mesh-contracts"), SeatOutsideItsDomainKind)
before = len(said)
// mesh-contracts defines wants and vault, seats of other domains: refused with a finding naming both modules, the
// held entries stand, and lone (Core) is taken.
@@ -472,7 +472,7 @@ func TestTheControlRecords(t *testing.T) {
// The lead's design limit (design 32): a merge names each package by its name in its own ecosystem, read at the
// merge commit: the Go package by its go.mod's module path, the TypeScript one by its package.json name.
files := map[string]string{"go/go.mod": "module " + sdkGo + "\n\ngo 1.26\n", "package.json": `{"name":"@novox/mesh-sdk","version":"0.0.0"}`}
ix := &Indexer{Packages: newMem(), Logf: t.Logf, ReadFile: func(_ context.Context, r, sha, p string) ([]byte, error) {
ix := &Indexer{Module: testSelf, Packages: newMem(), Logf: t.Logf, ReadFile: func(_ context.Context, r, sha, p string) ([]byte, error) {
if v, ok := files[p]; ok && r == "novox/mesh-sdk" && sha == "m1" {
return []byte(v), nil
}
modules/mesh-delivery-indexer/internal/index/entry_verb_test.go
+2 -2
View File
@@ -14,7 +14,7 @@ import (
func TestEntryFindsAnEntryByItsFingerprint(t *testing.T) {
ctx := context.Background()
fp := "src2:" + strings.Repeat("b", 64)
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
entries := map[string]meshcatalog.CatalogEntry{
"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}},
"none": {Module: "none", Repository: "novox/r", Commit: "c", Manifest: map[string]any{}},
@@ -48,7 +48,7 @@ func TestEntryFindsAnEntryByItsFingerprint(t *testing.T) {
func TestAKeptRevisionIndexesItsEntries(t *testing.T) {
ctx := context.Background()
fp := "src2:" + strings.Repeat("d", 64)
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
if err := ix.keepEntries(ctx, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}}}); err != nil {
t.Fatal(err)
}
modules/mesh-delivery-indexer/internal/index/findings.go
+1 -1
View File
@@ -33,7 +33,7 @@ func keyID(id string) string {
func (ix *Indexer) observed(scope, id, kind, summary, headline, explanation, source string) mcseat.Finding {
id = keyID(id)
return mcseat.Finding{Scope: scope, ID: id, Kind: kind, Severity: mcseat.FindingSeverityWarning, Summary: summary, Headline: headline,
Explanation: explanation, Source: source, Observer: Module, At: ix.now().UTC(), Subject: mcseat.Subject{Scope: scope, ID: id}}
Explanation: explanation, Source: source, Observer: ix.Module, At: ix.now().UTC(), Subject: mcseat.Subject{Scope: scope, ID: id}}
}
// clearFinding clears the finding of scope, id and kind when the writer holds it: a clear of none held says nothing.
modules/mesh-delivery-indexer/internal/index/helpers_test.go
+5 -1
View File
@@ -123,5 +123,9 @@ func (m *memFindings) Held() []mcseat.Finding {
// unsaidKey is the key of module's unsaid finding.
func unsaidKey(module string) string {
return views.FindingKey(mcc.ScopeModule, keyID(unsaidID(module)), UnsaidKind)
return views.FindingKey(mcc.ScopeModule, keyID(unsaidID(testSelf, module)), UnsaidKind)
}
// testSelf is the module the tests' indexer runs as, its credential's: its findings, its own entry and its graph key are named
// from it, never from a constant (design 47 G A-296).
const testSelf = "acme-indexer"
modules/mesh-delivery-indexer/internal/index/index_test.go
+1 -1
View File
@@ -133,7 +133,7 @@ func newIndexer(t *testing.T) (*Indexer, *memStore, *strings.Builder) {
_ = b.Put(buildRecord(t, "novox/mesh-delivery", "a", "modules/a", "2026-10-19T09:00:00Z"))
_ = b.Put(buildRecord(t, "novox/mesh-catalog", "c", "modules/c", "2026-10-19T09:00:00Z"))
_ = b.Put(buildRecord(t, "novox/mesh-delivery", "odd-name", "modules/odd", "2026-10-19T09:00:00Z"))
return &Indexer{Store: st, Builds: b, Logf: func(f string, a ...any) {
return &Indexer{Module: testSelf, Store: st, Builds: b, Logf: func(f string, a ...any) {
logs.WriteString(strings.TrimSpace(strings.ReplaceAll(f, "\n", " ")) + "\n")
_ = a
}}, st, logs
modules/mesh-delivery-indexer/internal/index/modules_test.go
+2 -2
View File
@@ -61,7 +61,7 @@ func record(t *testing.T, id, module, repo, path, ref, state, at, _ string) (str
func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) {
t.Helper()
mods := newMem()
ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()}
ix := &Indexer{Module: testSelf, Store: newMem(), Modules: mods, Builds: NewBuilds()}
ix.UseBuildModules(fixtureBuildModules)
for _, r := range recs {
if err := ix.Builds.Put(r[1].([]byte)); err != nil {
@@ -505,7 +505,7 @@ func TestAModuleMovesBetweenRepositoriesOnlyWhenItsTrunkDropsIt(t *testing.T) {
} {
t.Run(tc.name, func(t *testing.T) {
mods := &deletesSeen{memStore: newMem()}
ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()}
ix := &Indexer{Module: testSelf, Store: newMem(), Modules: mods, Builds: NewBuilds()}
var logs []string
ix.Logf = func(format string, args ...any) { logs = append(logs, format) }
if tc.heads {
modules/mesh-delivery-indexer/internal/index/names.go
-3
View File
@@ -4,9 +4,6 @@ package index
// `reads`) and from the manifests of the holders it reads; a unit row checks them against the manifest
// (TestTheNamesAreTheManifests). No contract names a holder (novox/hq ADR 0625 §4, ADR 0634 §1).
const (
// Module is the indexer's own name; each reader names it from its own manifest.
Module = "mesh-delivery-indexer"
// The indexer's own states, with the revisions each keeps per key (module.json `state`).
CandidatesState = "candidates"
CandidatesHistory = 10
modules/mesh-delivery-indexer/internal/index/open_asks_test.go
+4 -4
View File
@@ -22,7 +22,7 @@ func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) {
if err := pending.Create(ctx, "m.1", val); err != nil {
t.Fatal(err)
}
ix := &Indexer{Pending: pending}
ix := &Indexer{Module: testSelf, Pending: pending}
cancelled, cancel := context.WithCancel(ctx)
cancel()
for _, c := range []struct {
@@ -43,7 +43,7 @@ func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) {
}
})
t.Run("the seed's candidate and the versions' module are in the newest trunk revision", func(t *testing.T) {
ix := &Indexer{Revisions: newMemObjects(1 << 20)}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20)}
keep := func(n int64, fp string) {
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: n, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: fp, Manifest: map[string]any{}}})
if err == nil {
@@ -70,7 +70,7 @@ func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) {
})
t.Run("a branch is watched while a registration names it", func(t *testing.T) {
g := newRegRig()
ix := &Indexer{Registry: g.r}
ix := &Indexer{Module: testSelf, Registry: g.r}
raw, _ := json.Marshal(meshcatalog.Registration{Module: "avahi", Branches: []string{"main", "feat"}})
if err := g.r.Registrations.Create(ctx, "avahi", raw); err != nil {
t.Fatal(err)
@@ -91,7 +91,7 @@ func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) {
func TestAClosedAskIsNotUnanswered(t *testing.T) {
ctx := context.Background()
closed := fmt.Errorf("steps: %w", ErrAskClosed)
ix := &Indexer{Findings: newFindings(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Versions: newMem()}
ix := &Indexer{Module: testSelf, Findings: newFindings(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Versions: newMem()}
rev, objects, _ := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 1, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: "fp", Manifest: map[string]any{},
Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}}}})
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
modules/mesh-delivery-indexer/internal/index/pending.go
+4 -4
View File
@@ -916,7 +916,7 @@ func (ix *Indexer) dry() *dryRun {
return o, o
}
d := &dryRun{}
x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
x := &Indexer{Module: ix.Module, Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains,
ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil,
ValidationAsks: nil, AskVerb: nil, // a dry run asks no validation (§713)
@@ -988,7 +988,7 @@ func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (strin
if applyErr != nil {
return "", d, applyErr
}
key, _ := meshcatalogclient.ModulesKey(Module)
key, _ := meshcatalogclient.ModulesKey(ix.Module)
value, _, found, err := d.ix.Modules.Get(ctx, key)
if err != nil {
return "", d, err
@@ -1212,11 +1212,11 @@ func saidWaits(entries []PendingEntry) []takein.TakeIn {
// manifest, from the dry run; the manifests and the packages are written by the dry run itself. The revision is said
// pending under its oldest merge's key, and a wait it was said under before by another key is said taken.
func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own string, d *dryRun) error {
key, _ := meshcatalogclient.ModulesKey(Module)
key, _ := meshcatalogclient.ModulesKey(ix.Module)
if err := copyKey(ctx, d.modules, ix.Modules, key); err != nil {
return fmt.Errorf("the indexer's own module entry: %w", err)
}
if gk, err := graph.Key(Module); err == nil {
if gk, err := graph.Key(ix.Module); err == nil {
if err := copyKey(ctx, d.graph, ix.Graph, gk); err != nil && !errors.Is(err, process.ErrNotYet) {
return fmt.Errorf("the indexer's key of the module graph: %w", err)
}
modules/mesh-delivery-indexer/internal/index/pending_test.go
+32 -32
View File
@@ -41,7 +41,7 @@ type takeInBed struct {
func newTakeInBed(t *testing.T) *takeInBed {
t.Helper()
ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")},
mod{Module, renamed(t, "avahi", Module)})
mod{testSelf, renamed(t, "avahi", testSelf)})
if err := ix.Apply(context.Background(), f); err != nil {
t.Fatal(err)
}
@@ -65,7 +65,7 @@ func newTakeInBed(t *testing.T) *takeInBed {
logf(format, args...)
}
}
own := b.fp(t, Module)
own := b.fp(t, testSelf)
if own == "" || b.fp(t, "avahi") == "" {
t.Fatalf("the bed's modules have no fingerprint")
}
@@ -99,7 +99,7 @@ func (b *takeInBed) merge(number int, c, avahi, indexer string) forgemsg.Forge {
f.Trees["modules/avahi"] = avahi
}
if indexer != "" {
f.Trees["modules/"+Module] = indexer
f.Trees["modules/"+testSelf] = indexer
}
return f
}
@@ -228,14 +228,14 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b.arrive(t, 8, b.merge(31, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b.arrive(t, 9, b.merge(32, sha('d'), "tree-avahi-4", "tree-indexer-2"))
b.said = nil
b.arrive(t, 10, b.merge(33, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+Module]))
b.arrive(t, 10, b.merge(33, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+testSelf]))
if want := []string{"taken " + takein.PendingKey(8), "taken " + takein.PendingKey(9), "taken " + takein.PendingKey(10)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
t.Errorf("said %v, pending %v: want %v, the wait's key (m.8, said pending) once and each other entry's own", b.events(), b.keys(), want)
}
})
t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) {
b := newTakeInBed(t)
own, avahi0 := b.fp(t, Module), b.fp(t, "avahi")
own, avahi0 := b.fp(t, testSelf), b.fp(t, "avahi")
// Issue 1966 note 154 (ADR 0668 §4): with the verdicts state not on the bus yet, nothing is taken in and no manifest
// is written without its verdict; its bind evaluates pending again, which takes the merge in with its verdicts.
verdictsBound := b.ix.Verdicts
@@ -260,11 +260,11 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals)
// T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending.
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
named := b.fp(t, Module)
named := b.fp(t, testSelf)
if named == own || b.fp(t, "avahi") != avahi1 || b.commit(t, "avahi") != sha('b') {
t.Fatalf("indexer change: indexer %s (own %s), avahi %s at %s: want only the indexer written", named, own, b.fp(t, "avahi"), b.commit(t, "avahi"))
}
if g := graphOf(t, b.ix, Module); g.FP != named {
if g := graphOf(t, b.ix, testSelf); g.FP != named {
t.Errorf("graph of the indexer names %s, want %s", g.FP, named)
}
if n := len(b.ix.Verdicts.(*memStore).vals); n > verdicts+1 {
@@ -284,9 +284,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Fatalf("behind: pending %v, avahi at %s, said %v: want both pending and nothing said again", b.keys(), b.commit(t, "avahi"), b.events())
}
// A revert brings the indexer back to the running build: everything is taken in, oldest first.
b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+Module]))
if len(b.keys()) != 0 || b.fp(t, Module) != own || b.commit(t, "avahi") != sha('e') {
t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, Module), b.commit(t, "avahi"))
b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+testSelf]))
if len(b.keys()) != 0 || b.fp(t, testSelf) != own || b.commit(t, "avahi") != sha('e') {
t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, testSelf), b.commit(t, "avahi"))
}
if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) {
t.Errorf("said %v, want %v", b.events(), want)
@@ -309,7 +309,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
if err != nil || after.Revision != 2 {
t.Fatalf("after the merge naming a newer indexer the newest revision is %+v (%v), want revision 2", after, err)
}
if changed := changedEntries(before.Entries, after.Entries); !slices.Equal(changed, []string{Module}) {
if changed := changedEntries(before.Entries, after.Entries); !slices.Equal(changed, []string{testSelf}) {
t.Errorf("the revision differs from the previous in %v, want the indexer's entry alone", changed)
}
})
@@ -324,7 +324,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b.ix.Packages = newMem()
}
b.ix.GoMod = func(_ context.Context, _, commit, dir string) ([]byte, error) {
if commit == f.Merge && dir == "modules/"+Module {
if commit == f.Merge && dir == "modules/"+testSelf {
return []byte("module x\n\nreplace " + sibling + " => ../mesh-new/pkg\n"), nil
}
return []byte("module x\n"), nil
@@ -377,9 +377,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
for dir, raw := range b.base.Manifests {
f.Manifests[dir] = raw
}
dir := "modules/" + Module
f.Manifests[dir] = strings.Replace(f.Manifests[dir], `"module": "`+Module+`"`,
`"module": "`+Module+`", "invokes": ["seat:node-engine.`+verb+`"]`, 1)
dir := "modules/" + testSelf
f.Manifests[dir] = strings.Replace(f.Manifests[dir], `"module": "`+testSelf+`"`,
`"module": "`+testSelf+`", "invokes": ["seat:node-engine.`+verb+`"]`, 1)
if row.defined {
f.Trees[contracts] = "tree-seat-contracts"
f.Locks = map[string]map[string]string{}
@@ -423,11 +423,11 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
g.Repository, g.Trees, g.Manifests, g.Locks, g.Group = "novox/mesh-sdk", map[string]string{}, nil, nil, "feat/x"
g.Members = []meshdeliveryclient.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}}
b.arrive(t, 5, g)
named := b.fp(t, Module)
named := b.fp(t, testSelf)
// The new indexer starts; its build is not known (the engine does not answer at its start): a hand ask is answered
// meanwhile (R5), and the start waits for nothing. It is never asked again, so it stays unknown until it is started again.
mk := func(ask func(context.Context) (string, error)) *Indexer {
n := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
n := &Indexer{Module: testSelf, Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
n.UseBuildModules(b.ix.BuildModules())
n.toolchains = b.ix.toolchains
@@ -569,7 +569,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
b.ix.ownMu.Lock()
b.ix.ownFP = b.fp(t, Module)
b.ix.ownFP = b.fp(t, testSelf)
b.ix.ownMu.Unlock()
return b
}
@@ -693,7 +693,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) {
b := newTakeInBed(t)
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
named := b.fp(t, Module)
named := b.fp(t, testSelf)
b.ix.ownMu.Lock()
b.ix.ownFP = named
b.ix.ownMu.Unlock()
@@ -749,7 +749,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b := newTakeInBed(t)
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
named := b.fp(t, Module)
named := b.fp(t, testSelf)
b.ix.ownMu.Lock()
b.ix.ownFP = named
b.ix.ownMu.Unlock()
@@ -768,7 +768,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Run("W4.6 a full store holds the merge with the reason objects full, says full once, and room takes it in", func(t *testing.T) {
b := newTakeInBed(t)
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
named := b.fp(t, Module)
named := b.fp(t, testSelf)
b.ix.ownMu.Lock()
b.ix.ownFP = named
b.ix.ownMu.Unlock()
@@ -801,7 +801,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b2 := newTakeInBed(t)
b2.arrive(t, 2, b2.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b2.ix.ownMu.Lock()
b2.ix.ownFP = b2.fp(t, Module)
b2.ix.ownFP = b2.fp(t, testSelf)
b2.ix.ownMu.Unlock()
b2.ix.Revisions = newMemObjects(kept * 100 / 85)
var filling []meshcatalog.Fill
@@ -820,7 +820,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
b := newTakeInBed(t)
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
b.ix.ownMu.Lock()
b.ix.ownFP = b.fp(t, Module)
b.ix.ownFP = b.fp(t, testSelf)
b.ix.ownMu.Unlock()
b.ix.Revisions = store
return b
@@ -947,7 +947,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
answer = out.(takein.PendingAnswer)
}
}
if len(answer.Waits) != 1 || answer.Waits[0].Key != takein.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", Module}) ||
if len(answer.Waits) != 1 || answer.Waits[0].Key != takein.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{testSelf, "avahi"}) ||
answer.Waits[0].Said != takein.TakeInPendingEvent || len(answer.Waits[0].Keys) != 2 {
t.Fatalf("indexer_pending = %+v", answer)
}
@@ -1025,10 +1025,10 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
if len(b.keys()) != 2 || len(raw) > 4096 || b.commit(t, "avahi") != sha('b') {
t.Fatalf("a big merge held: pending %v, value %d bytes, avahi at %s: want two small values", b.keys(), len(raw), b.commit(t, "avahi"))
}
named := b.fp(t, Module)
named := b.fp(t, testSelf)
// The new indexer starts with nothing in memory and the bridge down: the newest merge is said lost with the bridge's
// words (event, state), the older one is covered and not lost, and nothing is taken in.
next := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
next := &Indexer{Module: testSelf, Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: bridgeDown, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf,
GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
next.UseBuildModules(b.ix.BuildModules())
@@ -1070,7 +1070,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
}
// A hand take-in of a merge by its commit asks the bridge and runs the message through Arrive.
b.ix.Message = answers
b.publish(t, big(b.merge(30, sha('e'), "tree-avahi-6", b.base.Trees["modules/"+Module])))
b.publish(t, big(b.merge(30, sha('e'), "tree-avahi-6", b.base.Trees["modules/"+testSelf])))
if r := b.ix.TakeInMerge(ctx, catalogueRepo, sha('e'), 9, "lead@shanks", "held by the cap"); !r.Stored || b.commit(t, "avahi") != sha('e') {
t.Fatalf("merge take-in: %+v, avahi at %s: want it taken in", r, b.commit(t, "avahi"))
}
@@ -1078,9 +1078,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Errorf("merge take-in of a merge the bridge does not know: %+v", r)
}
// An old value that holds the message whole reads as before.
old := b.merge(31, sha('f'), "tree-avahi-7", b.base.Trees["modules/"+Module])
old := b.merge(31, sha('f'), "tree-avahi-7", b.base.Trees["modules/"+testSelf])
rawOld, _ := forgemsg.EncodeForge(old)
data, _ := encodePending(Pending{Forge: rawOld, Since: b.ix.now(), LastWriter: b.fp(t, Module)})
data, _ := encodePending(Pending{Forge: rawOld, Since: b.ix.now(), LastWriter: b.fp(t, testSelf)})
if err := b.pending.Create(ctx, takein.PendingKey(11), data); err != nil {
t.Fatal(err)
}
@@ -1102,9 +1102,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
data, _ := takein.EncodeTaken(takein.Taken{Repository: trunk.RepositoryName(catalogueRepo), Merge: sha('d')})
key, _ := trunk.RepositoryKey(catalogueRepo)
b.ix.Taken.(*memStore).put(key, data)
named := b.fp(t, Module)
named := b.fp(t, testSelf)
asked := 0
next := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
next := &Indexer{Module: testSelf, Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Taken: b.ix.Taken, Contains: b.ix.Contains,
Message: func(context.Context, string, string, int) ([]byte, error) {
asked++
@@ -1128,7 +1128,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
t.Run("the dry run names every field of the indexer", func(t *testing.T) {
// dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails.
named := map[string]bool{}
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
for _, f := range []string{"Module", "Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged",
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn",
"Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "InputObjects", "BuildEdges", "Pins", "PinsPending", "ValidationAsks", "AskVerb"} {
modules/mesh-delivery-indexer/internal/index/pins.go
+3 -3
View File
@@ -235,14 +235,14 @@ func (ix *Indexer) unpin(ctx context.Context, key, record string) error {
return err
}
func pinFindingID(key string) string { return Module + ".pin." + key }
func pinFindingID(own, key string) string { return own + ".pin." + key }
func (ix *Indexer) sayPinMissing(ctx context.Context, key, record string) {
if ix.Findings == nil {
return
}
object, _ := pinObject(key)
f := ix.observed(mcc.ScopeModule, pinFindingID(key), PinMissingKind,
f := ix.observed(mcc.ScopeModule, pinFindingID(ix.Module, key), PinMissingKind,
fmt.Sprintf("the delivery record %s pins the catalog graph revision %s, which the indexer no longer holds", record, object),
"A delivery record points at a catalog graph revision the indexer does not keep",
"The indexer removed the revision, or never kept it, before the record pinned it. indexer_unpin removes the record's pin and clears this.", PinMissingKind)
@@ -255,7 +255,7 @@ func (ix *Indexer) clearPinMissing(ctx context.Context, key string) {
if ix.Findings == nil {
return
}
if err := ix.clearFinding(ctx, mcc.ScopeModule, pinFindingID(key), PinMissingKind, "no record pins "+key+" any more"); err != nil {
if err := ix.clearFinding(ctx, mcc.ScopeModule, pinFindingID(ix.Module, key), PinMissingKind, "no record pins "+key+" any more"); err != nil {
ix.logf("pins: the finding %s for %s is not cleared: %v", PinMissingKind, key, err)
}
}
modules/mesh-delivery-indexer/internal/index/pins_test.go
+3 -3
View File
@@ -18,7 +18,7 @@ func TestTheIndexerRemovesARevisionOnlyWhenNoRecordPinsItAndItIsNotCurrent(t *te
ctx := context.Background()
clock := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
pending := uint64(0)
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem(), Findings: newFindings(), Now: func() time.Time { return clock },
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), Pins: newMem(), Findings: newFindings(), Now: func() time.Time { return clock },
PinsPending: func(context.Context) (uint64, error) { return pending, nil }}
reader := meshcatalog.CatalogEntry{Module: "reader", Repository: "novox/r", Manifest: map[string]any{}}
for _, commit := range []string{"c1", "c2", "c3"} {
@@ -90,7 +90,7 @@ func TestTheIndexerRemovesARevisionOnlyWhenNoRecordPinsItAndItIsNotCurrent(t *te
func TestAPinsStatedBeforeTheStateWasMadeCountsNothing(t *testing.T) {
ctx := context.Background()
clock := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem(), Now: func() time.Time { return clock },
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), Pins: newMem(), Now: func() time.Time { return clock },
PinsPending: func(context.Context) (uint64, error) { return 0, nil }}
raw, _ := json.Marshal(meshdelivery.PinsStatedEvent{At: clock.Add(-time.Hour)})
if err := ix.HeardPins(ctx, meshdelivery.SubjectPinsStated, raw); err != nil {
@@ -103,7 +103,7 @@ func TestAPinsStatedBeforeTheStateWasMadeCountsNothing(t *testing.T) {
func TestHeadsReplacedDropsTheCandidatesGraph(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Revisions: newMemObjects(1 << 20), Pins: newMem()}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), Pins: newMem()}
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Kind: "candidate", Group: "feat-x", Heads: "0123456789ab", Revision: 1}, nil)
if err != nil {
t.Fatal(err)
modules/mesh-delivery-indexer/internal/index/registration_findings_test.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
func TestRegistrationFindings(t *testing.T) {
ctx := context.Background()
w := newFindings()
ix := &Indexer{Findings: w}
ix := &Indexer{Module: testSelf, Findings: w}
held := func(kind string) bool {
for _, f := range w.Held() {
if f.Scope == mcc.ScopeSeat && f.Kind == kind {
modules/mesh-delivery-indexer/internal/index/registrations.go
+3 -1
View File
@@ -110,6 +110,8 @@ func RegistrationKey(module string) string { return keys.EscapeName(module) }
// Registry is the registrations and the trusted changes of them. Its parts are the indexer's: the two states, the history
// stream, the event, the seat git's branch, the seat operator-channel's ask.
type Registry struct {
// Module is the indexer's own module, the credential's.
Module string
Registrations Store
Proposals Store
// History keeps one version on the stream registration-history, once under its key <module>.<version>; HistoryOf reads
@@ -725,7 +727,7 @@ func (r *Registry) Heard(ctx context.Context, w oc.Warrant) error {
if rec.Ask != w.Ask || rec.Option != w.Option || rec.AskDigest != w.AskDigest || rec.Asker != w.Asker {
return r.endKey(ctx, key, *p, ProposalRefused, "the warrant heard is not the one the router's record holds")
}
o, err := occ.For(*rec, Module, p.composeAsk(key))
o, err := occ.For(*rec, r.Module, p.composeAsk(key))
if err == nil {
err = occ.Performs(o, p.act())
}
modules/mesh-delivery-indexer/internal/index/registrations_test.go
+2 -2
View File
@@ -32,7 +32,7 @@ type regRig struct {
func newRegRig() *regRig {
g := &regRig{asks: map[string]oc.Ask{}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), unprotect: map[string]bool{}, chosen: map[string]string{}, missing: map[string]bool{}}
g.r = &Registry{Registrations: MemoryStore(), Proposals: MemoryStore(),
g.r = &Registry{Module: testSelf, Registrations: MemoryStore(), Proposals: MemoryStore(),
History: func(_ context.Context, r meshcatalog.Registration) error {
g.history = append(g.history, r)
return nil
@@ -71,7 +71,7 @@ func newRegRig() *regRig {
func (g *regRig) warrant(id, option string) oc.Warrant {
a := g.asks[id]
o, _ := occ.OptionOf(a, option)
return oc.Warrant{Ask: id, Asker: Module, Outcome: occ.OutcomeChosen, Option: option, Level: o.Level, Label: o.Label,
return oc.Warrant{Ask: id, Asker: testSelf, Outcome: occ.OutcomeChosen, Option: option, Level: o.Level, Label: o.Label,
AskDigest: occ.Digest(a), At: occ.FormatTime(g.now.Add(time.Minute)), By: &oc.Person{Who: occ.Operator, Kind: "telegram", Verified: "user id verified"}}
}
modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go
+1 -1
View File
@@ -73,7 +73,7 @@ func (l *lineMem) Subscribe(_ string, fn nats.MsgHandler) (*nats.Subscription, e
func TestAnInputOverOneMiBIsKeptAsAnObjectAndAnsweredInChunks(t *testing.T) {
ctx := context.Background()
objs := &objectsMem{objs: map[string][]byte{}}
ix := &Indexer{Inputs: newMem(), InputObjects: objs, Now: time.Now}
ix := &Indexer{Module: testSelf, Inputs: newMem(), InputObjects: objs, Now: time.Now}
body := make([]byte, 3<<20+17)
if _, err := rand.Read(body); err != nil {
t.Fatal(err)
modules/mesh-delivery-indexer/internal/index/runcandidates_test.go
+3 -3
View File
@@ -23,7 +23,7 @@ func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded(
ix, _, _ := manifestsIndexer(t)
cands, runs := newMem(), newMem()
ix.Store, ix.Runs, ix.Revisions = cands, runs, newMemObjects(0)
ix.Registry = &Registry{Registrations: MemoryStore()}
ix.Registry = &Registry{Module: testSelf, Registrations: MemoryStore()}
for _, r := range []meshcatalog.Registration{
{Module: "avahi", Origin: catalogueRepo, Path: "modules/avahi", Branches: []string{"main"}},
{Module: "netcheck", Origin: "novox/mesh-networking", Path: "modules/netcheck", Branches: []string{"main", "beta"}},
@@ -70,7 +70,7 @@ func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded(
t.Fatal(err)
}
// A restart: a fresh indexer over the same states.
again := &Indexer{Store: cands, Runs: runs, Revisions: ix.Revisions, Registry: ix.Registry, Read: ix.Read}
again := &Indexer{Module: testSelf, Store: cands, Runs: runs, Revisions: ix.Revisions, Registry: ix.Registry, Read: ix.Read}
if _, _, found, _ := cands.Get(ctx, key); !found {
t.Fatal("the candidate went before its last member's pipeline ended")
}
@@ -136,7 +136,7 @@ func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded(
func TestTheRunsAreReadAndDroppedByHand(t *testing.T) {
ctx := context.Background()
cands, runs := newMem(), newMem()
ix := &Indexer{Store: cands, Runs: runs, Now: func() time.Time { return time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) }}
ix := &Indexer{Module: testSelf, Store: cands, Runs: runs, Now: func() time.Time { return time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) }}
put := func(s *memStore, key string, v any) {
t.Helper()
data, _ := json.Marshal(v)
modules/mesh-delivery-indexer/internal/index/seatdefs.go
+9 -9
View File
@@ -458,7 +458,7 @@ func (ix *Indexer) twiceFinding(ctx context.Context, repository, module string,
lines = append(lines, fmt.Sprintf("%s: defined by %s (%s) and already by %s", seat, repository, module, holder))
}
slices.Sort(lines)
ix.finding(ctx, findingID("seat-defined-twice", module), SeatDefinedTwiceKind, len(lines) > 0,
ix.finding(ctx, findingID(ix.Module, "seat-defined-twice", module), SeatDefinedTwiceKind, len(lines) > 0,
fmt.Sprintf("%s's trunk defines seats another trunk already defines, refused: %s", module, strings.Join(lines, "; ")),
"A seat on "+module+"'s trunk is already defined by another repository",
"A seat has one definition: the one that stood first stays the seat's contract and "+module+"'s is kept nowhere. "+
@@ -473,7 +473,7 @@ func (ix *Indexer) outsideFinding(ctx context.Context, module string, outside ma
lines = append(lines, seat+": "+why)
}
slices.Sort(lines)
ix.finding(ctx, findingID("seat-outside-its-domain", module), SeatOutsideItsDomainKind, len(lines) > 0,
ix.finding(ctx, findingID(ix.Module, "seat-outside-its-domain", module), SeatOutsideItsDomainKind, len(lines) > 0,
fmt.Sprintf("%s's trunk defines seats of another domain, refused: %s", module, strings.Join(lines, "; ")),
"A seat on "+module+"'s trunk belongs to another domain's contracts module",
"A seat of the mesh's own contracts is taken only from its domain's contracts module (mesh-contracts for a domain with none). "+
@@ -510,8 +510,8 @@ func (ix *Indexer) sayChanged(ctx context.Context, c SeatDefinitionChanged) {
}
// findingID is the finding's subject id for a contracts module.
func findingID(what, module string) string {
return Module + "." + what + "." + module
func findingID(own, what, module string) string {
return own + "." + what + "." + module
}
// unparsedFinding writes the contracts module's seat-definition-unparsed finding for its trunk's seats that did not
@@ -522,7 +522,7 @@ func (ix *Indexer) unparsedFinding(ctx context.Context, module string, unparsed
lines = append(lines, seat+": "+why)
}
slices.Sort(lines)
ix.finding(ctx, findingID("seat-definitions", module), SeatDefinitionUnparsedKind, len(lines) > 0,
ix.finding(ctx, findingID(ix.Module, "seat-definitions", module), SeatDefinitionUnparsedKind, len(lines) > 0,
fmt.Sprintf("%s's trunk holds seat definitions that do not parse: %s", module, strings.Join(lines, "; ")),
"A seat definition on "+module+"'s trunk does not parse",
"The trunk's last definition that parsed stays the seat's contract. A change fixing the definition clears this when it "+
@@ -570,13 +570,13 @@ func (ix *Indexer) ApplyOutOfStepBuild(ctx context.Context, value []byte) {
case len(e.OutOfStep) > 0:
files := slices.Clone(e.OutOfStep)
slices.Sort(files)
ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, true,
ix.finding(ctx, findingID(ix.Module, "out-of-step", e.Module), OutOfStepKind, true,
fmt.Sprintf("%s's trunk build at %.12s has generated files out of step with their seat definitions: %s", e.Module, e.Commit, strings.Join(files, ", ")),
"Generated types on "+e.Module+"'s trunk are out of step with their seat definitions",
"Its build's test step regenerated the types and found the committed files differ, missing or not written by the "+
"generator (a generator's change does this). A change regenerating them clears this once a build of "+e.Module+" ends clean.")
case e.Outcome == builds.BuildDone:
ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, false, "", "", "")
ix.finding(ctx, findingID(ix.Module, "out-of-step", e.Module), OutOfStepKind, false, "", "", "")
}
}
@@ -676,7 +676,7 @@ type OutOfStepRow struct {
// OutOfStepOf answers indexer_out_of_step: every trunk contracts module whose finding generated-types-out-of-step
// stands, then every pull request whose check's verdict named out-of-step files.
func OutOfStepOf(defs, findings views.Source) ([]OutOfStepRow, error) {
func OutOfStepOf(own string, defs, findings views.Source) ([]OutOfStepRow, error) {
out := []OutOfStepRow{}
fkeys, err := findings.Keys()
if err != nil {
@@ -692,7 +692,7 @@ func OutOfStepOf(defs, findings views.Source) ([]OutOfStepRow, error) {
if err := json.Unmarshal(v.Value, &f); err != nil || f.Kind != OutOfStepKind {
continue
}
out = append(out, OutOfStepRow{Source: SourceTrunk, Module: strings.TrimPrefix(f.Subject.ID, Module+".out-of-step."), Summary: f.Summary})
out = append(out, OutOfStepRow{Source: SourceTrunk, Module: strings.TrimPrefix(f.Subject.ID, own+".out-of-step."), Summary: f.Summary})
}
keys, err := defs.Keys()
if err != nil {
modules/mesh-delivery-indexer/internal/index/seed_steps_test.go
+2 -2
View File
@@ -20,7 +20,7 @@ func TestTheSeedAsksStepsOfTheBuildSeatsOfTheTrunkEntries(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
w := newFindings()
ix := &Indexer{Findings: w, Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }}
ix := &Indexer{Module: testSelf, Findings: w, Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }}
fp := "src2:" + strings.Repeat("a", 64)
entries := map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c1", Path: "mods/avahi", Fp: fp, Manifest: map[string]any{},
Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}, {Ref: "avahi:image", Language: "go", Kind: "image"}},
@@ -78,7 +78,7 @@ func TestTheSeedAsksStepsOfTheBuildSeatsOfTheTrunkEntries(t *testing.T) {
t.Fatal("a call unanswered for no time wrote its finding")
}
now = now.Add(5 * time.Minute)
ix = &Indexer{Findings: w, Heard: ix.Heard, Revisions: ix.Revisions, Now: func() time.Time { return now }} // a restart keeps the time
ix = &Indexer{Module: testSelf, Findings: w, Heard: ix.Heard, Revisions: ix.Revisions, Now: func() time.Time { return now }} // a restart keeps the time
if _, err := ix.SeedModulesFrom(ctx, defs, call); err != nil {
t.Fatal(err)
}
modules/mesh-delivery-indexer/internal/index/toolchains_test.go
+7 -7
View File
@@ -606,7 +606,7 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { //
verbObjects := newMemObjects(0)
keep := func(rev meshcatalog.CatalogRevision, declares map[string]bool) {
entries := map[string][]byte{}
for _, name := range []string{"avahi", Module, "mesh-grants", "node-tools"} {
for _, name := range []string{"avahi", testSelf, "mesh-grants", "node-tools"} {
e := meshcatalog.CatalogEntry{Module: name, Repository: catalogueRepo}
if declares[name] {
e.Invokes = []string{RevisionInvoke}
@@ -620,22 +620,22 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { //
t.Fatal(err)
}
}
if got, err := Revision(context.Background(), verbObjects, Module, "n1.node-tools", meshcatalog.RevisionInput{}); err != nil || !got.NoneYet || got.Revision != nil || len(got.Entries) != 0 {
if got, err := Revision(context.Background(), verbObjects, testSelf, "n1.node-tools", meshcatalog.RevisionInput{}); err != nil || !got.NoneYet || got.Revision != nil || len(got.Entries) != 0 {
t.Errorf("revision with none kept answered %+v, %v, want none-yet alone with no error (design 47 G §530)", got, err)
}
keep(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, map[string]bool{"node-tools": true, Module: true})
keep(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, map[string]bool{"node-tools": true, testSelf: true})
keep(meshcatalog.CatalogRevision{Revision: 4, Kind: "candidate", Group: "feat/x", Heads: "0123456789ab"}, map[string]bool{"node-tools": true, "mesh-grants": true})
for caller, answered := range map[string]bool{"n1.node-tools": true, "n1.mesh-grants": false, "n1." + Module: false, "person.someone": false, "unproven": false} {
got, err := Revision(context.Background(), verbObjects, Module, caller, meshcatalog.RevisionInput{})
for caller, answered := range map[string]bool{"n1.node-tools": true, "n1.mesh-grants": false, "n1." + testSelf: false, "person.someone": false, "unproven": false} {
got, err := Revision(context.Background(), verbObjects, testSelf, caller, meshcatalog.RevisionInput{})
if (err == nil) != answered || (answered && (got.Revision.Revision != 3 || len(got.Entries) != 4 || got.Next != "")) {
t.Errorf("revision to %s answered %+v (%v), want answered %v with trunk revision 3 whole", caller, got, err, answered)
}
}
if got, err := Revision(context.Background(), verbObjects, Module, "n1.node-tools", meshcatalog.RevisionInput{Group: "feat/x", Heads: "0123456789ab", After: "mesh-grants"}); err != nil ||
if got, err := Revision(context.Background(), verbObjects, testSelf, "n1.node-tools", meshcatalog.RevisionInput{Group: "feat/x", Heads: "0123456789ab", After: "mesh-grants"}); err != nil ||
got.Revision.Kind != "candidate" || len(got.Entries) != 1 || got.Entries[0].Module != "node-tools" {
t.Errorf("a candidate's revision after mesh-grants answered %+v (%v), want its one entry node-tools", got, err)
}
if _, err := Revision(context.Background(), verbObjects, Module, "n1.node-tools", meshcatalog.RevisionInput{Revision: 9}); err == nil {
if _, err := Revision(context.Background(), verbObjects, testSelf, "n1.node-tools", meshcatalog.RevisionInput{Revision: 9}); err == nil {
t.Error("a trunk revision not kept was answered")
}
// ADR 0614 §2, §3: the graph names what each module invokes and serves, and leaves out, over the whole graph, the
modules/mesh-delivery-indexer/internal/index/validate_test.go
+1 -1
View File
@@ -87,7 +87,7 @@ func TestValidationSeatsAreTheRevisionsWithTheMergesOver(t *testing.T) {
}
v, _ := json.Marshal(SeatDefinitionEntry{Seat: "pulled", Source: SourcePull, Pull: 3, Digest: "d", Definition: json.RawMessage(`{"seat":"pulled"}`)})
defs.put(SeatDefinitionKey("pulled", SourcePull, "r", 3), v)
ix := &Indexer{SeatDefinitions: defs}
ix := &Indexer{Module: testSelf, SeatDefinitions: defs}
seatsOf := func(raws []json.RawMessage) map[string]string {
out := map[string]string{}
for _, raw := range raws {
modules/mesh-delivery-indexer/internal/index/verdicts.go
+1 -1
View File
@@ -571,7 +571,7 @@ func (ix *Indexer) FixOwnBuild(ctx context.Context) {
// trunk holds: the indexer's own modules value names it. A newer build merged and not yet delivered ends it at the trunk's
// merge, as one that pins a changed mesh-sdk always is; false when either is unknown.
func (ix *Indexer) ownBuildIsNewest(held map[string]meshcatalog.ModulesByModuleValue) bool {
fp := held[Module].Fp2
fp := held[ix.Module].Fp2
if fp == "" {
return false
}
modules/mesh-delivery-indexer/internal/index/verdicts_test.go
+3 -3
View File
@@ -23,7 +23,7 @@ import (
// mended clears the module's refusal (novox/hq ADR 0578 §1).
func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1",
ix := &Indexer{Module: testSelf, Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1",
Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
var said []meshcatalog.ManifestRefusedEvent
ix.SayRefused = func(_ context.Context, r meshcatalog.ManifestRefusedEvent) { said = append(said, r) }
@@ -427,7 +427,7 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) {
// event) while a newer build of the indexer is on the trunk, and is judged again at the next judge; once this program's
// build is the trunk's newest it is refused; a wait is written into the module's value as a refusal naming the fields.
ownFP := "src2:" + strings.Repeat("e", 64)
hold(Module, `{"module":"mesh-delivery-indexer","version":"1.0"}`)
hold(testSelf, `{"module":"mesh-delivery-indexer","version":"1.0"}`)
ix.ownFP = ownFP // this program started when the trunk held another build of the indexer than the one it holds now
later := hold("later", `{"module":"later","version":"1.0","a-field-from-later":1,"another":{"x":1}}`)
said = nil
@@ -570,7 +570,7 @@ func (unreadable) Get(context.Context, string) ([]byte, uint64, bool, error) {
// each naming both; a module with its own name is accepted.
func TestAPackageNameBelongsToOneModule(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1",
ix := &Indexer{Module: testSelf, Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1",
Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }}
ix.ReadFile = func(_ context.Context, _, _, path string) ([]byte, error) {
if strings.HasSuffix(path, "go.mod") {
modules/mesh-delivery-indexer/internal/index/versions_test.go
+8 -8
View File
@@ -25,7 +25,7 @@ func TestAVersionIsReleasedOnceEveryArtifactIsPublished(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
var said []meshcatalog.VersionReleasedEvent
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
SayVersionReleased: func(_ context.Context, e meshcatalog.VersionReleasedEvent, id string) error {
said = append(said, e)
return nil
@@ -95,7 +95,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
w := newFindings()
var filledSaid []meshcatalog.VersionsFilledEvent
ix := &Indexer{Versions: newMem(), Heard: newMem(), Findings: w, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), Findings: w, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now },
SayVersionsFilled: func(_ context.Context, e meshcatalog.VersionsFilledEvent, id string) error {
filledSaid = append(filledSaid, e)
return nil
@@ -171,7 +171,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
// makes nothing behind.
func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
entries := map[string]meshcatalog.CatalogEntry{
"lib": {Module: "lib", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "lib:pkg"}}},
"dep": {Module: "dep", Manifest: map[string]any{}, Artifacts: []meshcatalog.Artifact{{Ref: "dep:program"}}},
@@ -255,7 +255,7 @@ func TestOutdatedAnswersTheEdgesBehindByTheirPartsDigest(t *testing.T) {
func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }}
ix := &Indexer{Module: testSelf, Inputs: newMem(), Now: func() time.Time { return now }}
d1, err := ix.KeepInput(ctx, "ask-1", []byte("first"))
if err != nil {
t.Fatal(err)
@@ -303,7 +303,7 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) {
// name parses and nothing of it is written to the own-build state.
func TestWhatIsHeardIsKeptEscapedInTheStateHeard(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), OwnBuildState: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
ix := &Indexer{Module: testSelf, Versions: newMem(), OwnBuildState: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
if err := ix.HearVersionDelivered(ctx, meshwants.VersionDeliveredEvent{Module: "a.b", Version: "1.2.3"}); err != nil {
t.Fatal(err)
}
@@ -328,7 +328,7 @@ func TestWhatIsHeardIsKeptEscapedInTheStateHeard(t *testing.T) {
// released), and a kind-6 edge when the delivered version's entry no longer serves the word.
func TestOutdatedAnswersKinds3And5And6(t *testing.T) {
ctx := context.Background()
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20)}
keep := func(n, source int64, serves []string, fp string) {
t.Helper()
entries := map[string]meshcatalog.CatalogEntry{
@@ -433,7 +433,7 @@ func TestTheFirstFillGivesTheNewestVersionsTheirFilledEdges(t *testing.T) {
ctx := context.Background()
t0 := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
placed := map[string][]string{"prov": {"n1", "n2"}}
ix := &Indexer{Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return t0 },
ix := &Indexer{Module: testSelf, Versions: newMem(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Now: func() time.Time { return t0 },
Placement: func(_ context.Context, m string) ([]meshmoduleassigner.ModulePlacement, error) {
return []meshmoduleassigner.ModulePlacement{{Branch: "trunk", Nodes: placed[m]}}, nil
},
@@ -501,7 +501,7 @@ func TestTheFirstFillGivesTheNewestVersionsTheirFilledEdges(t *testing.T) {
func TestAnInputIsDeletedAtTheStepEndedOfItsAsk(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }}
ix := &Indexer{Module: testSelf, Inputs: newMem(), Now: func() time.Time { return now }}
d1, err := ix.KeepInput(ctx, "a.1", []byte("first"))
if err != nil {
t.Fatal(err)
modules/mesh-delivery-indexer/internal/index/wanted_names_test.go
+2 -2
View File
@@ -100,7 +100,7 @@ func TestAnEntryNoRevisionListsStaysWhileAWantNamesItAndGoesAfterTheWantMovesOn(
ctx := context.Background()
fp := namedFp("c")
n := NewNamed(newMem(), nil)
ix := &Indexer{Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), NamedSets: n, Now: func() time.Time { return namedBase }}
ix := &Indexer{Module: testSelf, Revisions: newMemObjects(1 << 20), EntryIndex: newMem(), NamedSets: n, Now: func() time.Time { return namedBase }}
n.Changed = ix.RemoveUnlisted
if err := ix.keepEntries(ctx, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c", Fp: fp, Manifest: map[string]any{}}}); err != nil {
t.Fatal(err)
@@ -173,7 +173,7 @@ func TestNamedDropRemovesAMemberByHandAndEntriesReadsTheSet(t *testing.T) {
n := NewNamed(st, nil)
namedEvent{wanted: true, node: "n1", module: "m", fp: fp, min: 1}.apply(t, n)
namedEvent{wanted: true, node: "n2", module: "m", fp: fp, min: 1}.apply(t, n)
ix := &Indexer{NamedSets: n, Logf: func(string, ...any) {}}
ix := &Indexer{Module: testSelf, NamedSets: n, Logf: func(string, ...any) {}}
var acts []string
asked := func(key string, value []byte) (TakeInAsk, error) {
a, err := DecodeTakeIn(value)
modules/mesh-delivery-indexer/manifest_test.go
+1 -4
View File
@@ -144,9 +144,6 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
// D-11 (ADR 0634 §1): the holders and buckets the indexer names (internal/index/names.go) are its manifest's: its own
// name and states with their revisions, and the states it reads, each as `<holder>.<state>`.
if index.Module != m.Module {
t.Errorf("the indexer names itself %q, its manifest's module is %q", index.Module, m.Module)
}
// A-296: its own buckets are named from its credential's module (index.Bucket), each a state its manifest keeps.
own := func(state string) string { return index.Bucket("a-indexer", state) }
for bucket, history := range map[string]int{own(index.CandidatesState): index.CandidatesHistory, own(index.ManifestsState): index.ManifestsHistory,
@@ -214,7 +211,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
// Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section.
if len(m.Streams) != 2 || m.Streams[0].Name != index.RegistrationHistoryStream || m.Streams[0].Retention != "history" || m.Streams[0].MaxMiB != 64 ||
m.Streams[1].Name != index.Module+"-findings" || m.Streams[1].MaxAgeDays != 7 || m.Streams[1].MaxMiB != 16 {
m.Streams[1].Name != m.Module+"-findings" || m.Streams[1].MaxAgeDays != 7 || m.Streams[1].MaxMiB != 16 {
t.Errorf("streams = %+v, want registration-history keeping its history, 64 MiB, and the findings stream (§387)", m.Streams)
}
// Design 47 G §368: version is MAJOR.MINOR (0.1 while none is published, §368); §295, §299: the catalog revisions' form and