event hub: a start that found no durable on the bus is tried again at the next start, so a part is not lost for the order in which the bus makes its consumer (B7e; issue 1966)

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

go/process/event_hub.go

@@ -19,11 +19,9 @@ import (
 type EventHub struct {
 	logf func(string, ...any)
 
-	mu       sync.RWMutex
-	routes   []hubRoute
-	started  bool
-	cons     jetstream.ConsumeContext
-	startErr error
+	mu     sync.RWMutex
+	routes []hubRoute
+	cons   jetstream.ConsumeContext
 }
 
 type hubRoute struct {
@@ -68,23 +66,23 @@ func (h *EventHub) Patterns() []string {
 	return out
 }
 
-// Start binds the module's durable on the shared events stream and pulls it; later calls do nothing and return the
-// first one's result. It makes no stream and no consumer.
+// Start binds the module's durable on the shared events stream and pulls it; a call after a start that took does
+// nothing. It makes no stream and no consumer: a consumer the bus does not hold yet is the error of this call, and a
+// later call (the module's next change of membership) binds it, so a part is not lost for the bus's order.
 func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Account) error {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	if h.started {
-		return h.startErr
+	if h.cons != nil {
+		return nil
 	}
-	h.started = true
 	cons, err := cred.EventConsumer(ctx, js, EventsStream)
 	if err == nil {
 		h.cons, err = HearEvents(cons, h.hear, h.logf)
 	}
 	if err != nil {
-		h.startErr = fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
+		return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
 	}
-	return h.startErr
+	return nil
 }
 
 // Stop ends the pull.

go/views/process_reader_integration_test.go

@@ -156,4 +156,31 @@ func TestTheViewsReaderAndTwoHandlersShareTheModulesOneDurable(t *testing.T) {
 	if info, err := cons.Info(ctx); err != nil || info.NumAckPending != 0 {
 		t.Fatalf("a message was left unacknowledged: %+v %v", info, err)
 	}
+	// A hub started before the bus holds its durable says so, and binds at a later start: a module is not lost for the
+	// order in which the bus makes its consumer.
+	late := process.Account{Node: "n1", Module: "late"}
+	lateHub := process.NewEventHub(t.Logf)
+	heard := make(chan string, 1)
+	if err := lateHub.Handle("mesh.mod.l.event.>", func(m jetstream.Msg) { _ = m.Ack(); heard <- m.Subject() }); err != nil {
+		t.Fatal(err)
+	}
+	defer lateHub.Stop()
+	if err := lateHub.Start(ctx, js, late); err == nil {
+		t.Fatal("a hub with no durable on the bus started")
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: late.EventConsumerName(),
+		AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{"mesh.mod.l.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := lateHub.Start(ctx, js, late); err != nil {
+		t.Fatalf("the hub did not bind at its second start: %v", err)
+	}
+	if _, err := js.Publish(ctx, "mesh.mod.l.event.x", []byte("{}")); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-heard:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the hub started late heard nothing")
+	}
 }