From 8c685160eab3a10f287a395cf982b605d214c259 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 16:12:05 +0100 Subject: [PATCH] =?UTF-8?q?Indexer:=20its=20reader=20of=20its=20own=20seat?= =?UTF-8?q?=20list=20reads=20again=20as=20the=20states=20it=20answers=20fr?= =?UTF-8?q?om=20are=20bound,=20with=20no=20definition=20moving=20(ADR=2006?= =?UTF-8?q?36=20=C2=A71;=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-indexer/main.go | 3 +- .../cmd/mesh-delivery-indexer/seatlist.go | 19 ++++++ .../mesh-delivery-indexer/seatlist_test.go | 66 +++++++++++++++++++ 3 files changed, 87 insertions(+), 1 deletion(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.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 f2a48b7f..558a874f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -408,7 +408,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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) + seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) + readWhenBound(seatReader, seatDefs, marks) 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, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go index 28ea31a4..508c0cac 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go @@ -129,3 +129,22 @@ func followSeatList(bus *seatBus, say func(context.Context, index.SeatDefinition return nil } } + +// bucket is a state bound when it appears (process.AwaitedBucket). +type bucket interface{ Ready(func(jetstream.KeyValue)) } + +// readWhenBound has the reader read the list again as each of the states it is answered from is bound. The reader starts before +// they are, its first read finds them missing and leaves it unread, and the indexer's own events (said) are the only other thing +// that reads it again: a trunk that moves no definition after a start would leave the list unread for good (ADR 0636 §1). +func readWhenBound(reader *seatlist.Reader, states ...bucket) { + if reader == nil { + return + } + for _, state := range states { + state.Ready(func(jetstream.KeyValue) { + if !reader.Status().Heard { + go func() { _ = reader.Reload() }() + } + }) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go new file mode 100644 index 00000000..268bf0de --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" +) + +type unboundSource struct { + mu sync.Mutex + bound bool +} + +func (s *unboundSource) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + if !s.bound { + return nil, errors.New("not yet") + } + return nil, nil +} + +func (s *unboundSource) Get(string) (*views.Entry, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + if !s.bound { + return nil, false, errors.New("not yet") + } + return nil, false, nil +} + +type awaitedFake struct{ ready []func(jetstream.KeyValue) } + +func (b *awaitedFake) Ready(f func(jetstream.KeyValue)) { b.ready = append(b.ready, f) } + +// The indexer's reader of its own seat list starts before its states are bound, so its first read finds them missing; it reads the +// list again when they are bound, with no definition moving (ADR 0636 §1). +func TestTheSeatListIsReadAgainWhenItsStatesAreBound(t *testing.T) { + src := &unboundSource{} + bus := &seatBus{defs: src, marks: src} + reader, _ := followSeatList(bus, nil, t.Logf) + <-reader.Ready() + if reader.Status().Heard { + t.Fatal("the list was read with its states missing") + } + b := &awaitedFake{} + readWhenBound(reader, b) + src.mu.Lock() + src.bound = true + src.mu.Unlock() + for _, f := range b.ready { + f(nil) + } + for deadline := time.Now().Add(5 * time.Second); !reader.Status().Heard; time.Sleep(10 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the list was not read after its states were bound") + } + } + var _ seatlist.Bus = bus +}