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
2026-11-05 15:13:12 +01:00
parent c16b38bd83
commit 07a75a768a
3 changed files with 149 additions and 2 deletions
@@ -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),
@@ -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)
@@ -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
+	}
+}