Merge branch 'feat/47-G-fix11-b2f' into feat/47-G-join11

go/process/event_hub_integration_test.go

@@ -4,6 +4,7 @@ package process_test
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -192,9 +193,89 @@ func subAModuleHearsAnotherModulesStreamKeptLastPerSubject(t *testing.T) {
 	}
 }
 
+// A-233: a module hears a stream kept by age with FromFirst: at every start the stream is read whole from its first message, though
+// its durable began after them, then the durable is followed; without FromFirst only the durable is.
+func subAModuleHearsAStreamKeptByAgeFromItsFirstMessage(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()
+	kept, subjects := seatdef.KeptStream("mesh-catalog", seatdef.Message{Name: "version-released"}), []string{"mesh.seat.mesh-catalog.event.version-released"}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxAge: 90 * 24 * time.Hour}); err != nil {
+		t.Fatal(err)
+	}
+	for _, v := range []string{"web 1.0.0", "web 1.1.0", "api 2.0.0"} {
+		if _, err := js.Publish(ctx, subjects[0], []byte(v)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cred := process.Account{Node: "ace", Module: "assigner"}
+	durable := cred.SeatEventConsumerConfig(subjects, false)
+	durable.DeliverPolicy = jetstream.DeliverNewPolicy
+	if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil {
+		t.Fatal(err)
+	}
+	got := make(chan string, 8)
+	hear := func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() }
+	var reads []string
+	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.FromFirst(), process.OnRead(func(s string) { reads = append(reads, s) }))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var heard []string
+	for len(heard) < 3 {
+		select {
+		case h := <-got:
+			heard = append(heard, h)
+		case <-ctx.Done():
+			t.Fatalf("heard %v, want every message from the first", heard)
+		}
+	}
+	if strings.Join(heard, ",") != "web 1.0.0,web 1.1.0,api 2.0.0" || len(reads) != 2 {
+		t.Errorf("heard %v with reads %v, want every message in order, read whole", heard, reads)
+	}
+	if _, err := js.Publish(ctx, subjects[0], []byte("web 1.2.0")); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case h := <-got:
+		if h != "web 1.2.0" {
+			t.Errorf("the durable heard %s", h)
+		}
+	case <-ctx.Done():
+		t.Fatal("the durable was not followed after the read")
+	}
+	cc.Stop()
+	// Without FromFirst, a stream kept by age is not read: the durable alone, past every message, hands the next one first.
+	cc, err = cred.HearSeatEvents(ctx, js, kept, hear, t.Logf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cc.Stop()
+	if _, err := js.Publish(ctx, subjects[0], []byte("next")); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case h := <-got:
+		if h != "next" {
+			t.Errorf("without FromFirst %s was handed before the next message", h)
+		}
+	case <-ctx.Done():
+		t.Fatal("the durable was not followed")
+	}
+}
+
 // TestTheHubBindsALateConsumerAndSaysItIsLive runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
 func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
 	t.Run("TheHubBindsALateConsumerAndSaysItIsLive", subTheHubBindsALateConsumerAndSaysItIsLive)
 	t.Run("AModuleHearsASeatEventKeptOnItsOwnStream", subAModuleHearsASeatEventKeptOnItsOwnStream)
 	t.Run("AModuleHearsAnotherModulesStreamKeptLastPerSubject", subAModuleHearsAnotherModulesStreamKeptLastPerSubject)
+	t.Run("AModuleHearsAStreamKeptByAgeFromItsFirstMessage", subAModuleHearsAStreamKeptByAgeFromItsFirstMessage)
 }

go/process/seatstream.go

@@ -46,17 +46,26 @@ const (
 // HearOption is an option of HearSeatEvents.
 type HearOption func(*hearOptions)
 
-type hearOptions struct{ onRead func(state string) }
+type hearOptions struct {
+	onRead    func(state string)
+	fromFirst bool
+}
 
 // OnRead is the callback HearSeatEvents calls with ReadWaiting and then ReadDone around its start read; the module wires it
 // to its first views (design 47 G A-19), so this package imports nothing of views.
 func OnRead(f func(state string)) HearOption { return func(o *hearOptions) { o.onRead = f } }
 
+// FromFirst is HearStream's form for a stream kept by age and cap, not last per subject (design 47 G A-233, A-189): at every
+// start the stream is read whole from its first message through an ordered consumer made for that one read, on the subjects of
+// the durable, and only then is the durable followed. A message the durable has not acknowledged yet is so handed twice, once
+// by the read and once by the durable: the handler takes a message again as it took it.
+func FromFirst() HearOption { return func(o *hearOptions) { o.fromFirst = true } }
+
 // readWhole hands handle every message an ordered consumer, made on stream for the one read, delivers from the last of each
-// subject filtered to subjects, until none is left pending (design 47 G §556): it ends when the message it delivered has
+// subject (or, from DeliverAll, from the first message, FromFirst) filtered to subjects, until none is left pending (design 47 G §556): it ends when the message it delivered has
 // none after it, or at once when the consumer, made, holds none.
-func readWhole(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, handle func(jetstream.Msg)) error {
-	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: subjects, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
+func readWhole(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, handle func(jetstream.Msg), from jetstream.DeliverPolicy) error {
+	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: subjects, DeliverPolicy: from})
 	if err != nil {
 		return err
 	}
@@ -100,9 +109,22 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str
 	return a.HearStream(ctx, js, stream, handle, logf, opts...)
 }
 
+// wholeRead is how a start read begins: from the last of each subject for a stream kept last per subject, from the first
+// message for FromFirst's, and none for another stream.
+func wholeRead(maxPerSubject int64, fromFirst bool) (jetstream.DeliverPolicy, bool) {
+	switch {
+	case maxPerSubject == 1:
+		return jetstream.DeliverLastPerSubjectPolicy, true
+	case fromFirst:
+		return jetstream.DeliverAllPolicy, true
+	}
+	return 0, false
+}
+
 // HearStream is HearSeatEvents for the stream of a module's own events (a manifest's `streams` entry, named as
 // catalogue.StreamBusName names it), which the module is granted through its `consumes`: the same one rule, a stream the bus's own
-// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181).
+// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181);
+// with FromFirst, a stream kept by age is read whole from its first message the same way (A-233).
 func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) {
 	if handle == nil {
 		return nil, fmt.Errorf("hearing %s needs a handler", stream)
@@ -122,7 +144,7 @@ func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream
 	if err != nil {
 		return nil, fmt.Errorf("the stream %s is not read: %w", stream, err)
 	}
-	if st.CachedInfo().Config.MaxMsgsPerSubject == 1 {
+	if from, whole := wholeRead(st.CachedInfo().Config.MaxMsgsPerSubject, o.fromFirst); whole {
 		if o.onRead != nil {
 			o.onRead(ReadWaiting)
 		}
@@ -131,7 +153,7 @@ func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream
 		if cfg.FilterSubject != "" {
 			subjects = append(subjects, cfg.FilterSubject)
 		}
-		if err := readWhole(ctx, js, stream, subjects, handle); err != nil {
+		if err := readWhole(ctx, js, stream, subjects, handle, from); err != nil {
 			return nil, fmt.Errorf("the start read of %s is not made: %w", stream, err)
 		}
 		if o.onRead != nil {