Indexer: its reader of its own seat list reads again as the states it answers from are bound, with no definition moving (ADR 0636 §1; issue 1966)
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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() }()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user