seat list: ReloadIfUnread reads the list again once when a host's events path is live; event hub: StartUntilBound binds a late consumer without a restart (B9e; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

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()

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

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

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