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 + } +}