From 5595403215ecda453c00f9bda8a4ce00ce7d3216 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:10:12 +0100 Subject: [PATCH] The indexer opens its catalog graph revisions under the module its account runs as, not a fixed name (milestone 47/G W4.2) A module's name in its own code logic is a fixed name the runtime already says: the bus account the process runs as names it. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../mesh-delivery-indexer/integration_test.go | 8 +++---- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/main_test.go | 7 ++++++ .../cmd/mesh-delivery-indexer/objects.go | 23 ++++++++++++------- 4 files changed, 27 insertions(+), 13 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 16f3ec69..e1d58357 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -481,7 +481,7 @@ var acceptedHeard atomic.Int64 type revisionView struct{ js jetstream.JetStream } func (v revisionView) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { - view, _, _, err := index.TrunkView(ctx, &busObjects{js: v.js, local: index.CatalogRevisionsObjects}) + view, _, _, err := index.TrunkView(ctx, newBusObjects(v.js, testAccount, index.CatalogRevisionsObjects)) if err != nil { return nil, err } @@ -497,7 +497,7 @@ func (v revisionView) Get(ctx context.Context, key string) (jetstream.KeyValueEn // History is the module's value in every trunk revision the objects keep that holds it, oldest first. func (v revisionView) History(ctx context.Context, key string) ([]jetstream.KeyValueEntry, error) { - o := &busObjects{js: v.js, local: index.CatalogRevisionsObjects} + o := newBusObjects(v.js, testAccount, index.CatalogRevisionsObjects) names, err := o.Names(ctx) if err != nil { return nil, err @@ -600,7 +600,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . t.Fatal(err) } // The indexer's objects catalog-revisions, which the grants part creates from its manifest's old objects section. - if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.ObjectsBucket(indexerModule, index.CatalogRevisionsObjects)}); err != nil { + if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: newBusObjects(nil, testAccount, index.CatalogRevisionsObjects).bucket()}); err != nil { t.Fatal(err) } modules := revisionView{js: js} @@ -1043,7 +1043,7 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC if err != nil { t.Fatal(err) } - if err := index.KeepRevision(ctx, &busObjects{js: js, local: index.CatalogRevisionsObjects}, rev, objects); err != nil { + if err := index.KeepRevision(ctx, newBusObjects(js, testAccount, index.CatalogRevisionsObjects), rev, objects); err != nil { t.Fatal(err) } kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.ToolchainsBucket("mesh-build-go"), History: 1}) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 610725e7..9faf2d03 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -378,7 +378,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The view of the trunk is the newest trunk revision of the catalog graph the objects keep (milestone 47/G, ADR 0656 §6: // one form of the fact), read once here and kept as a revision after every write; objects not on the bus yet, or none // kept, start it empty, and the first start's catch-up takes the trunk in once the gate's source is read (catchUp). - revisions := &busObjects{js: js, local: index.CatalogRevisionsObjects} + revisions := newBusObjects(js, cred, index.CatalogRevisionsObjects) modules, graph, at, err := index.TrunkView(ctx, revisions) switch { case errors.Is(err, process.ErrNotYet): diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 1450b161..11264cd3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -79,6 +79,9 @@ 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"} + 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). if em := viewsEmitter(context.Background(), nil, process.Account{}, nil, &process.AwaitedBucket{}, nil, t.Logf); em.Findings == nil { @@ -143,6 +146,10 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) t.Errorf("answered leaves %v, presence asked %v", held.vals, asked) } }) + // Milestone 47/G: the object buckets are the module's own by the name its account runs as, never a fixed name. + if b := newBusObjects(nil, process.Account{Module: "renamed-indexer"}, index.CatalogRevisionsObjects).bucket(); b != "renamed-indexer_catalog-revisions" { + t.Errorf("the catalog graph revisions open as %s, want the account's module's", b) + } t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered) var logs bytes.Buffer st := &memStore{vals: map[string][]byte{}} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go index bb6ae703..025301db 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go @@ -20,26 +20,33 @@ import ( "mesh-delivery-indexer/internal/index" ) -// indexerModule is the module this program runs as, whose object buckets it opens by their local names. -const indexerModule = "mesh-delivery-indexer" - // busObjects is one of the indexer's own object buckets in the old `objects` section (novox/hq ADR 0619 §2), which the // grants part creates: it is opened at its first use and kept; until the bucket exists every call is an error saying so, // and the next call opens it again (no timer: the next revision or tool call asks). type busObjects struct { - js jetstream.JetStream - local string - mu sync.Mutex - o jetstream.ObjectStore + js jetstream.JetStream + // module is the module this program runs as, read from its bus account (process.Account.Module), never a fixed name. + module string + local string + mu sync.Mutex + o jetstream.ObjectStore } +// newBusObjects is the object bucket of this local name of the module the account runs as. +func newBusObjects(js jetstream.JetStream, cred process.Account, local string) *busObjects { + return &busObjects{js: js, module: cred.Module, local: local} +} + +// bucket is the bus's name of the object bucket: the module's own name and the local name. +func (b *busObjects) bucket() string { return catalogue.ObjectsBucket(b.module, b.local) } + func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) { b.mu.Lock() defer b.mu.Unlock() if b.o != nil { return b.o, nil } - o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(indexerModule, b.local)) + o, err := b.js.ObjectStore(ctx, b.bucket()) if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { return nil, fmt.Errorf("the object bucket %s is not on the bus yet (the grants part creates it from the manifest's objects): %w", b.local, process.ErrNotYet) }