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