diff --git a/go/process/event_hub.go b/go/process/event_hub.go index f648d65..df8537e 100644 --- a/go/process/event_hub.go +++ b/go/process/event_hub.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" "sync" + "time" "github.com/nats-io/nats.go/jetstream" ) @@ -85,6 +86,44 @@ func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Accou return nil } +// StartUntilBound is Start that does not give up: when the module's consumer is not there yet it returns that error at +// once, and tries again on a goroutine of its own, after each wait, until it binds or ctx ends; then live is called (nil +// calls nothing), so a part that needs the events path live does what it does only then. wait returns false to stop +// trying (nil waits five seconds or for ctx). The bus makes the consumer when the module's account is composed, which +// says nothing a module can hear before the consumer exists, so the retry is a wait on that, bounded by ctx. +func (h *EventHub) StartUntilBound(ctx context.Context, js jetstream.JetStream, cred Account, wait func(context.Context) bool, live func()) error { + err := h.Start(ctx, js, cred) + if err == nil { + if live != nil { + live() + } + return nil + } + if wait == nil { + wait = func(ctx context.Context) bool { + t := time.NewTimer(5 * time.Second) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } + } + } + go func() { + for wait(ctx) && ctx.Err() == nil { + if h.Start(ctx, js, cred) == nil { + if live != nil { + live() + } + return + } + } + }() + return err +} + // Stop ends the pull. func (h *EventHub) Stop() { h.mu.Lock() diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go new file mode 100644 index 0000000..f8fa9bf --- /dev/null +++ b/go/process/event_hub_integration_test.go @@ -0,0 +1,67 @@ +//go:build integration + +package process_test + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// A module whose consumer the bus makes after the module started is not lost for that order: the hub tries again after +// each wait and calls live once it is bound (the checker's "until it starts again" hole). +func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { + t.Fatal(err) + } + cred := process.Account{Node: "n1", Module: "late"} + hub := process.NewEventHub(t.Logf) + waits, live := make(chan chan bool), make(chan struct{}, 1) + wait := func(ctx context.Context) bool { + step := make(chan bool) + select { + case waits <- step: + case <-ctx.Done(): + return false + } + return <-step + } + if err := hub.StartUntilBound(ctx, js, cred, wait, func() { live <- struct{}{} }); err == nil { + t.Fatal("no consumer was there, and the start said nothing") + } + select { + case <-live: + t.Fatal("live before the consumer was there") + default: + } + step := <-waits + step <- true // tried again: still no consumer + step = <-waits + if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: cred.EventConsumerName(), AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + step <- true + select { + case <-live: + case <-ctx.Done(): + t.Fatal("the consumer was made and the hub never said it is live") + } + hub.Stop() +} diff --git a/go/seatlist/seatlist.go b/go/seatlist/seatlist.go index 9d0dd06..cc366f3 100644 --- a/go/seatlist/seatlist.go +++ b/go/seatlist/seatlist.go @@ -115,6 +115,7 @@ type Reader struct { ready chan struct{} // closed once the first read of the seat has come to an end, answered or not revision uint64 lastErr string + again chan struct{} // the read ReloadIfUnread began, closed when it has come to an end } // Held is one definition a reader holds: its seat, where from, and the digest of the trunk file it came from. @@ -326,6 +327,37 @@ func (r *Reader) Reload() error { return nil } +// ReloadIfUnread reads the list again, on a goroutine of its own, when the seat has not answered yet; it is for a host whose +// path for the seat's events has just become live. The seat may have said it speaks before that path existed, and events +// start at the stream's end, so without this call the list would stay unread until the seat's next event (ADR 0636 ยง1). +// It returns nil when the list is already read; otherwise a channel closed when the read has come to an end, which is the +// reader's first read or the read a call before began while one is still running, never a second one beside it. Whether the +// seat answered is Status().Heard; an unanswered read leaves the list unread, to be read at the seat's next event. +func (r *Reader) ReloadIfUnread() <-chan struct{} { + r.mu.Lock() + defer r.mu.Unlock() + if r.spoken { + return nil + } + if !r.isReady() { + return r.ready + } + if r.again != nil { + select { + case <-r.again: + default: + return r.again + } + } + done := make(chan struct{}) + r.again = done + go func() { + _ = r.Reload() + close(done) + }() + return done +} + // fetch asks the seat for one definition and holds it, or drops it when the seat holds none. func (r *Reader) fetch(seat string) error { raw, err := r.cfg.Bus.Ask(Key(), map[string]any{"seat": seat}) diff --git a/go/seatlist/seatlist_test.go b/go/seatlist/seatlist_test.go index e196f4d..e0e8147 100644 --- a/go/seatlist/seatlist_test.go +++ b/go/seatlist/seatlist_test.go @@ -286,3 +286,43 @@ func TestTheBootstrapEqualsTheTestsSeatList(t *testing.T) { } } } + +// A host whose path for the seat's events became live after the seat spoke reads the list at once, once: a read +// already running is the one waited for, and a list that is read is left alone. +func TestReloadIfUnreadReadsAListTheSeatSaidItsEventBeforeTheHostHeardIt(t *testing.T) { + owner := &fakeOwner{down: true} + r, err := Start(Config{Bus: owner, Bootstrap: Bootstrapped()}) + if err != nil { + t.Fatal(err) + } + <-r.Ready() + if r.Status().Heard { + t.Fatal("the silent seat must leave the reader unheard") + } + // the seat still silent: a read is made and the list stays unread + if done := r.ReloadIfUnread(); done == nil { + t.Fatal("an unread list was not read again") + } else { + <-done + } + if r.Status().Heard { + t.Fatal("a silent seat cannot have been heard") + } + // the seat answered (its event was said before the host listened): the next call reads, and only one read runs beside another + owner.down = false + owner.table = map[string]string{"node-b": def("node-b", "node", "put")} + owner.gate = make(chan struct{}) + first := r.ReloadIfUnread() + second := r.ReloadIfUnread() + if first == nil || first != second { + t.Fatal("a second call beside a running read must wait for the same one") + } + close(owner.gate) + <-first + if st := r.Status(); !st.Heard || !r.Has("node-b") { + t.Fatalf("after the read: %+v", st) + } + if r.ReloadIfUnread() != nil { + t.Fatal("a list that is read was read again") + } +}