From 07a75a768a6a6cb78748c08337475686b92d2944 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:13:12 +0100 Subject: [PATCH] Indexer: it reads its own seat list as every part does (the seat list reader over its own state, its own moves heard at once) and gives catalogue what it holds, so the judge judges against the seats the mesh defines (D-29 host wiring; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 7 +- .../packages_integration_test.go | 13 ++ .../cmd/mesh-delivery-indexer/seatlist.go | 131 ++++++++++++++++++ 3 files changed, 149 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go 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 61ad6c36..b114ded7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -405,13 +405,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf) // The seat definitions (design 26, DLV-I17): a state this delivery brings, bound when it appears; until then none is kept. seatDefs := process.AwaitBucket(ctx, js, conn, index.SeatDefinitionsBucket, "no seat definition is kept", logger.Printf) + marks := process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf) + seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}} + _, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred), Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{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{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}, - SeatDefinitions: awaitedStore{seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeatDefinition(js, cred), + SeatDefinitions: awaitedStore{seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Message: messageVia(calls.Call), // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. @@ -643,7 +646,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro heard := awaitedStore{process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was // said up to; a first view (the tools bundle) answers at that mark. - startViews(ctx, js, cred, process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf), + startViews(ctx, js, cred, marks, findings, map[string]readyOf{"manifests": boundNow(manifests), "modules": boundNow(modules), "candidates": boundNow(candidates), index.VerdictsState: awaited(verdicts), delivery.IndexerPredefinedState: awaited(predefined), diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go index 4e32d014..58b6b254 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go @@ -16,6 +16,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/seatdef" @@ -392,6 +393,18 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T) t.Errorf("seat-definitions = %s, %v; want %s", raw, err, want) } } + // D-29: the indexer reads its own seat list as every part does, so the judge judges against the seats it said: the + // handed-over ones are known to catalogue (the test's seat list holds neither). + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(20 * time.Millisecond) { + _, wants := catalogue.SeatNamed("wants") + _, vault := catalogue.SeatNamed("vault") + if wants && vault { + break + } + if time.Now().After(deadline) { + t.Fatalf("the indexer's judge does not know the seats it said: wants %v, vault %v", wants, vault) + } + } // Said: git, wants and vault changed once each beyond the first count; no removal for the handed-over seats. if m, err := info.GetLastMsgForSubject(ctx, "mesh.seat.mesh-catalog.event.seat-definition-removed"); err == nil { t.Errorf("a handover said seat-definition-removed: %s", m.Data) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go new file mode 100644 index 00000000..28ea31a4 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go @@ -0,0 +1,131 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" + + "mesh-delivery-indexer/internal/index" +) + +// awaitedSource is a state the indexer owns, bound when it appears, as a views.Source: nothing is read before it is bound. +type awaitedSource struct { + ctx context.Context + b *process.AwaitedBucket +} + +func (s awaitedSource) Keys() ([]string, error) { + kv := s.b.KV() + if kv == nil { + return nil, process.ErrNotYet + } + k, err := kv.Keys(s.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return k, err +} + +func (s awaitedSource) Get(key string) (*views.Entry, bool, error) { + kv := s.b.KV() + if kv == nil { + return nil, false, process.ErrNotYet + } + e, err := kv.Get(s.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return &views.Entry{Value: e.Value(), Revision: e.Revision(), Written: e.Created()}, true, nil +} + +// The indexer holds the seat mesh-catalog, which serves the seat list (ADR 0636 §4), and judges every manifest against it: it +// reads its own list as every part does (the seat list reader), from the bootstrapped definitions until its state is read, and +// follows its own events, handed straight from where it says them. What the reader holds is given to catalogue at every +// change, which is the list the judge judges against (judge.UseSeatList). + +// seatBus is the reader's bus: the verb seat-definitions is answered from the indexer's own state and its two events are the +// ones the indexer says (said), never a call over the bus to itself. +type seatBus struct { + defs, marks views.Source + + mu sync.Mutex + handlers map[string][]func(string, json.RawMessage) error // by pattern +} + +func (b *seatBus) Ask(key string, body any) (json.RawMessage, error) { + if key != seatlist.Key() { + return nil, fmt.Errorf("the indexer's own seat list answers %s alone, not %s", seatlist.Key(), key) + } + var seat string + switch v := body.(type) { + case map[string]any: + seat, _ = v["seat"].(string) + case map[string]string: + seat = v["seat"] + } + answer, err := index.SeatDefinitionsOf(b.defs, b.marks, seat) + if err != nil { + return nil, err + } + return json.Marshal(answer) +} + +func (b *seatBus) Subscribe(pattern string, handle func(string, json.RawMessage) error) error { + b.mu.Lock() + defer b.mu.Unlock() + if b.handlers == nil { + b.handlers = map[string][]func(string, json.RawMessage) error{} + } + b.handlers[pattern] = append(b.handlers[pattern], handle) + return nil +} + +// said hands a definition's move, which the indexer has just said on the seat's subject, to the reader. +func (b *seatBus) said(c index.SeatDefinitionChanged) { + event, body := seatlist.ChangedEvent, map[string]string{"seat": c.Seat, "source": c.Source, "digest": c.Digest} + if c.Digest == "" { + event, body = seatlist.RemovedEvent, map[string]string{"seat": c.Seat} + } + data, _ := json.Marshal(body) + b.mu.Lock() + hs := append([]func(string, json.RawMessage) error(nil), b.handlers[seatlist.Pattern(event)]...) + b.mu.Unlock() + for _, h := range hs { + _ = h(seatlist.Pattern(event), data) + } +} + +// followSeatList starts the indexer's reader of its own seat list and gives catalogue what it holds at every change, and wraps +// say so that each move the indexer says is also heard by the reader. It returns the reader, whose status the indexer's health +// can show. +func followSeatList(bus *seatBus, say func(context.Context, index.SeatDefinitionChanged) error, logf func(string, ...any)) (*seatlist.Reader, func(context.Context, index.SeatDefinitionChanged) error) { + reader, err := seatlist.Start(seatlist.Config{Bus: bus, Bootstrap: seatlist.Bootstrapped(), + OnChange: func(seats []catalogue.Seat) { catalogue.UseSeats(seats) }}) + if err != nil { + logf("the indexer cannot follow its seat list: %v", err) + } + if reader != nil { + catalogue.UseSeats(reader.Seats()) + } + return reader, func(ctx context.Context, c index.SeatDefinitionChanged) error { + if err := say(ctx, c); err != nil { + return err + } + if c.Source == index.SourceTrunk { // only a trunk's definition is the seat list's + bus.said(c) + } + return nil + } +}